implemented DISTINCT (fb-1562) (#2388)

* implemented distinct

* implemented distinct
* uses first cut of a buffer pool, and extendible hashing with thresholded spill to disk
* tests
* cleaned up some stuff around query plan output to make developing tooling easier
* added optimization to call PQL Distinct()

* fixed test

* fix for passing wrong index name in orchestrator

* back out change to DistinctTimestamp

* fix other instance of wrong table name being passed

* use full index name instead of abbreviated one for translation. sigh.

* removed some unused code

Co-authored-by: Matthew Jaffee <jaffee@pilosa.com>
(cherry picked from commit f030d58d95)

bufferpool/bufferpool.go

@@ -0,0 +1,262 @@
+package bufferpool
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+)
+
+// FrameID is the type for frame id
+type FrameID int
+
+// PageID is the type for page id
+type PageID int
+
+var pageSyncPool = sync.Pool{
+	New: func() any {
+		pg := new(Page)
+		pg.id = PageID(INVALID_PAGE)
+		pg.isDirty = false
+		pg.pinCount = 0
+		return pg
+	},
+}
+
+// BufferPool represents a buffer pool of pages
+type BufferPool struct {
+	// the underlying storage
+	diskManager DiskManager
+	// the actual pages in the buffer pool
+	pages []*Page
+	// the replacer that will elect replacements when buffer pool is full
+	replacer *ClockReplacer
+	// the list of free frames
+	freeList []FrameID
+	// the map of frames to page ids to frame ids
+	// frame ids are the offset into pages
+	// if you ask the pool for page 673, this will know at
+	// what offset in pages page 673 will exist
+	pageTable map[PageID]FrameID
+}
+
+// TODO(pok) implement a lazy writer
+//  * if free list is 'low' then
+//		* increase size of cache if there is physical memory available
+//		* write out old pages and boot them from the cache to increase free list
+
+// TODO(pok) implement a checkpoint that scans the pool and writes out dirty pages every
+// minute or so
+
+// NewBufferPool returns a buffer pool
+func NewBufferPool(maxSize int, diskManager DiskManager) *BufferPool {
+	freeList := make([]FrameID, 0)
+	pages := make([]*Page, maxSize)
+	for i := 0; i < maxSize; i++ {
+		frameNumber := FrameID(i)
+		freeList = append(freeList, frameNumber)
+	}
+	clockReplacer := NewClockReplacer(maxSize)
+	return &BufferPool{
+		diskManager: diskManager,
+		pages:       pages,
+		replacer:    clockReplacer,
+		freeList:    freeList,
+		pageTable:   make(map[PageID]FrameID),
+	}
+}
+
+// Dumps all the pages in the buffer pool
+func (b *BufferPool) Dump() {
+	fmt.Println()
+	fmt.Printf("------------------------------------------------------------------------------------------\n")
+	fmt.Printf("BUFFER POOL\n")
+	for _, p := range b.pages {
+		if p != nil {
+			p.Dump("")
+		}
+	}
+	fmt.Printf("------------------------------------------------------------------------------------------\n")
+	fmt.Println()
+}
+
+// FetchPage fetches the requested page from the buffer pool.
+func (b *BufferPool) FetchPage(pageID PageID) (*Page, error) {
+	// if it is in buffer pool already then just return it
+	if frameID, ok := b.pageTable[pageID]; ok {
+		page := b.pages[frameID]
+		page.pinCount++
+		b.replacer.Pin(frameID)
+		return page, nil
+	}
+
+	// not in the buffer pool so try the free list or
+	// the replacer will vote a page off the island
+	frameID, isFromFreeList, err := b.getFrameID()
+	if err != nil {
+		return nil, err
+	}
+
+	if !isFromFreeList {
+		// if it didn't come from the freelist then
+		// remove page from current frame, writing it out if dirty
+		currentPage := b.pages[frameID]
+		if currentPage != nil {
+			if currentPage.isDirty {
+				b.diskManager.WritePage(currentPage)
+			}
+
+			delete(b.pageTable, currentPage.id)
+		}
+	}
+
+	// if we got to here, sorry, have to do an I/O
+	page, err := b.diskManager.ReadPage(pageID)
+	if err != nil {
+		return nil, err
+	}
+	page.pinCount = 1
+	b.pageTable[pageID] = frameID
+	pageSyncPool.Put(b.pages[frameID])
+	b.pages[frameID] = page
+	b.replacer.Pin(frameID)
+
+	return page, nil
+}
+
+// UnpinPage unpins the target page from the buffer pool
+func (b *BufferPool) UnpinPage(pageID PageID) error {
+	if frameID, ok := b.pageTable[pageID]; ok {
+		page := b.pages[frameID]
+		page.DecPinCount()
+
+		if page.pinCount <= 0 {
+			b.replacer.Unpin(frameID)
+		}
+		return nil
+	}
+
+	return errors.New("could not find page")
+}
+
+// FlushPage Flushes the target page to disk
+func (b *BufferPool) FlushPage(pageID PageID) bool {
+	if frameID, ok := b.pageTable[pageID]; ok {
+		page := b.pages[frameID]
+		page.DecPinCount()
+
+		b.diskManager.WritePage(page)
+		page.isDirty = false
+
+		return true
+	}
+	return false
+}
+
+// NewPage allocates a new page in the buffer pool with the disk manager help
+func (b *BufferPool) NewPage() (*Page, error) {
+	// get a free frame
+	frameID, isFromFreeList, err := b.getFrameID()
+	if err != nil {
+		return nil, err
+	}
+
+	if !isFromFreeList {
+		// remove page from current frame
+		currentPage := b.pages[frameID]
+		if currentPage != nil {
+			if currentPage.isDirty {
+				b.diskManager.WritePage(currentPage)
+			}
+
+			delete(b.pageTable, currentPage.id)
+		}
+	}
+
+	// allocates new page
+	pageID, err := b.diskManager.AllocatePage()
+	if err != nil {
+		return nil, err
+	}
+	page := &Page{pageID, 1, false, [PAGE_SIZE]byte{}}
+	page.WritePageNumber(int32(pageID))
+	page.WriteFreeSpaceOffset(int16(PAGE_SIZE))
+	page.WriteNextPointer(int32(INVALID_PAGE))
+	page.WritePrevPointer(int32(INVALID_PAGE))
+
+	// update the frame table
+	b.pageTable[pageID] = frameID
+	pageSyncPool.Put(b.pages[frameID])
+	b.pages[frameID] = page
+
+	return page, nil
+}
+
+// ScratchPage returns a page outside the buffer pool - do not use if you intend the page
+// to be in the buffer pool (use NewPage() for that)
+// ScratchPage is intended to be used in cases where you need the Page primitives
+// and will copy the scratch page back over a real page later
+func (b *BufferPool) ScratchPage() *Page {
+	page := &Page{
+		id:       PageID(INVALID_PAGE),
+		pinCount: 0,
+		isDirty:  false,
+		data:     [PAGE_SIZE]byte{},
+	}
+	page.WritePageNumber(int32(INVALID_PAGE))
+	page.WriteFreeSpaceOffset(int16(PAGE_SIZE))
+	page.WriteNextPointer(int32(INVALID_PAGE))
+	page.WritePrevPointer(int32(INVALID_PAGE))
+	return page
+}
+
+// DeletePage deletes a page from the buffer pool
+func (b *BufferPool) DeletePage(pageID PageID) error {
+	var frameID FrameID
+	var ok bool
+	if frameID, ok = b.pageTable[pageID]; !ok {
+		return nil
+	}
+
+	page := b.pages[frameID]
+
+	if page.pinCount > 0 {
+		return errors.New("pin count greater than 0")
+	}
+	delete(b.pageTable, page.id)
+	b.replacer.Pin(frameID)
+	b.diskManager.DeallocatePage(pageID)
+
+	b.freeList = append(b.freeList, frameID)
+
+	return nil
+}
+
+// FlushAllpages flushes all the pages in the buffer pool to disk
+// Yeah, never call this unless you know what you are doing
+func (b *BufferPool) FlushAllpages() {
+	for pageID := range b.pageTable {
+		b.FlushPage(pageID)
+	}
+}
+
+func (b *BufferPool) getFrameID() (FrameID, bool, error) {
+	if len(b.freeList) > 0 {
+		frameID, newFreeList := b.freeList[0], b.freeList[1:]
+		b.freeList = newFreeList
+		return frameID, true, nil
+	}
+
+	victim, err := b.replacer.Victim()
+	return victim, false, err
+}
+
+// OnDiskSize exposes the on disk size of the backing store
+// behind this buffer pool
+func (b *BufferPool) OnDiskSize() int64 {
+	return b.diskManager.FileSize()
+}
+
+// Close closes the buffer pool
+func (b *BufferPool) Close() {
+	b.diskManager.Close()
+}

bufferpool/circularlist.go

@@ -0,0 +1,93 @@
+package bufferpool
+
+import (
+	"errors"
+)
+
+type circularListNode struct {
+	key   interface{}
+	value interface{}
+	next  *circularListNode
+	prev  *circularListNode
+}
+
+type circularList struct {
+	head     *circularListNode
+	tail     *circularListNode
+	size     int
+	capacity int
+}
+
+func newCircularList(maxSize int) *circularList {
+	return &circularList{nil, nil, 0, maxSize}
+}
+
+func (c *circularList) find(key interface{}) *circularListNode {
+	ptr := c.head
+	for i := 0; i < c.size; i++ {
+		if ptr.key == key {
+			return ptr
+		}
+		ptr = ptr.next
+	}
+	return nil
+}
+
+func (c *circularList) hasKey(key interface{}) bool {
+	return c.find(key) != nil
+}
+
+func (c *circularList) insert(key interface{}, value interface{}) error {
+	if c.size == c.capacity {
+		return errors.New("list is full")
+	}
+	newNode := &circularListNode{key, value, nil, nil}
+	if c.size == 0 {
+		newNode.next = newNode
+		newNode.prev = newNode
+		c.head = newNode
+		c.tail = newNode
+		c.size++
+		return nil
+	}
+
+	node := c.find(key)
+	if node != nil {
+		node.value = value
+		return nil
+	}
+
+	newNode.next = c.head
+	newNode.prev = c.tail
+	c.tail.next = newNode
+	if c.head == c.tail {
+		c.head.next = newNode
+	}
+	c.tail = newNode
+	c.head.prev = c.tail
+
+	c.size++
+	return nil
+}
+
+func (c *circularList) remove(key interface{}) {
+	node := c.find(key)
+	if node == nil {
+		return
+	}
+	if c.size == 1 {
+		c.head = nil
+		c.tail = nil
+		c.size--
+		return
+	}
+	if node == c.head {
+		c.head = c.head.next
+	}
+	if node == c.tail {
+		c.tail = c.tail.prev
+	}
+	node.next.prev = node.prev
+	node.prev.next = node.next
+	c.size--
+}

bufferpool/clockreplacer.go

@@ -0,0 +1,64 @@
+package bufferpool
+
+import "errors"
+
+// ClockReplacer implements a clock replacer algorithm
+type ClockReplacer struct {
+	cList     *circularList
+	clockHand **circularListNode
+}
+
+// NewClockReplacer instantiates a new clock replacer
+func NewClockReplacer(poolSize int) *ClockReplacer {
+	cList := newCircularList(poolSize)
+	return &ClockReplacer{cList, &cList.head}
+}
+
+// Victim removes the victim frame as defined by the replacement policy
+func (c *ClockReplacer) Victim() (FrameID, error) {
+	if c.cList.size == 0 {
+		return FrameID(INVALID_PAGE), errors.New("no victims available")
+	}
+	var victimFrameID FrameID
+	currentNode := (*c.clockHand)
+	for {
+
+		if currentNode.value.(bool) {
+			currentNode.value = false
+			c.clockHand = &currentNode.next
+		} else {
+			frameID := currentNode.key.(FrameID)
+			victimFrameID = frameID
+			c.clockHand = &currentNode.next
+			c.cList.remove(currentNode.key)
+			return victimFrameID, nil
+		}
+	}
+}
+
+// Unpin unpins a frame, indicating that it can now be victimized
+func (c *ClockReplacer) Unpin(id FrameID) {
+	if !c.cList.hasKey(id) {
+		c.cList.insert(id, true)
+		if c.cList.size == 1 {
+			c.clockHand = &c.cList.head
+		}
+	}
+}
+
+// Pin pins a frame, indicating that it should not be victimized until it is unpinned
+func (c *ClockReplacer) Pin(id FrameID) {
+	node := c.cList.find(id)
+	if node == nil {
+		return
+	}
+	if (*c.clockHand) == node {
+		c.clockHand = &(*c.clockHand).next
+	}
+	c.cList.remove(id)
+}
+
+// Size returns the size of the clock
+func (c *ClockReplacer) Size() int {
+	return c.cList.size
+}

bufferpool/diskmanager.go

@@ -0,0 +1,21 @@
+package bufferpool
+
+// DiskManager is responsible for interacting with disk
+type DiskManager interface {
+	// reads a page from the disk
+	ReadPage(PageID) (*Page, error)
+	// writes a page to the disk
+	WritePage(*Page) error
+
+	// allocates a page
+	AllocatePage() (PageID, error)
+
+	// deallocates a page
+	DeallocatePage(PageID) error
+
+	// returns on disk file size
+	FileSize() int64
+
+	// closes and does any clean up
+	Close()
+}

bufferpool/inmemdiskmanager.go

@@ -0,0 +1,162 @@
+package bufferpool
+
+import (
+	"errors"
+	"fmt"
+	"os"
+
+	uuid "github.com/satori/go.uuid"
+)
+
+// InMemDiskSpillingDiskManager is a memory implementation for a DiskManager interface
+// that can spill to disk when a threshold is reached
+type InMemDiskSpillingDiskManager struct {
+	// tracks the number of pages
+	numPages int
+
+	onDiskPages int
+
+	// tracks the number of pages we can consume before spilling
+	thresholdPages int
+	hasSpilled     *struct{}
+	fd             *os.File
+
+	// the data buffer
+	data []byte
+}
+
+// NewInMemDiskSpillingDiskManager returns a in-memory version of disk manager
+func NewInMemDiskSpillingDiskManager(thresholdPages int) *InMemDiskSpillingDiskManager {
+	dm := &InMemDiskSpillingDiskManager{
+		numPages:       0,
+		thresholdPages: thresholdPages,
+		data:           make([]byte, 0),
+	}
+	return dm
+}
+
+// ReadPage reads a page from pages
+func (d *InMemDiskSpillingDiskManager) ReadPage(pageID PageID) (*Page, error) {
+	// check we're not asking for page out of range
+	if pageID < 0 || int(pageID) >= d.numPages {
+		return nil, errors.New("page not found")
+	}
+	// check that the offset is within range
+	offset := int(pageID) * PAGE_SIZE
+
+	var page = pageSyncPool.Get().(*Page)
+	// we have to do this stupid check because if -cpuprofile is set for go test, this
+	// the previous line return a weird nil-ish thing...
+	if page == (*Page)(nil) {
+		page = pageSyncPool.New().(*Page)
+	}
+	page.id = pageID
+
+	// do the read
+	if d.hasSpilled == nil {
+		if offset+PAGE_SIZE > len(d.data) {
+			return nil, errors.New("offset out of range")
+		}
+		b := copy(page.data[:], d.data[offset:offset+PAGE_SIZE])
+		fmt.Printf("bytes read: %d", b)
+	} else {
+		var err error
+		if offset+PAGE_SIZE > d.numPages*PAGE_SIZE {
+			return nil, errors.New("offset out of range")
+		}
+		_, err = d.fd.ReadAt(page.data[:], int64(offset))
+		if err != nil {
+			return nil, err
+		}
+	}
+	return page, nil
+}
+
+// WritePage writes a page in memory to pages
+func (d *InMemDiskSpillingDiskManager) WritePage(page *Page) error {
+	// make sure the offset is sensible
+	offset := int(page.ID()) * PAGE_SIZE
+	// do the write
+	if d.hasSpilled == nil {
+		if offset+PAGE_SIZE > len(d.data) {
+			return errors.New("offset out of range")
+		}
+		copy(d.data[offset:], page.data[:])
+	} else {
+		var err error
+		if offset+PAGE_SIZE > d.numPages*PAGE_SIZE {
+			return errors.New("offset out of range")
+		}
+		_, err = d.fd.WriteAt(page.data[:], int64(offset))
+		if err != nil {
+			return err
+		}
+		// err = d.fd.Sync()
+		// if err != nil {
+		// 	return err
+		// }
+	}
+	return nil
+}
+
+// AllocatePage allocates a page and returns the page number
+func (d *InMemDiskSpillingDiskManager) AllocatePage() (PageID, error) {
+	d.numPages = d.numPages + 1
+	pageID := PageID(d.numPages - 1)
+
+	if d.hasSpilled == nil {
+		// we have not spilled (yet), so make storage bigger
+		newData := make([]byte, PAGE_SIZE)
+		d.data = append(d.data, newData...)
+
+		// check to see if we need to spill
+		if d.numPages > d.thresholdPages {
+			fileUUID, err := uuid.NewV4()
+			if err != nil {
+				return PageID(INVALID_PAGE), err
+			}
+			// TODO(pok) we should try to tell the OS not to cache this file
+			d.fd, err = os.CreateTemp("", fmt.Sprintf("fb-ehash-%s", fileUUID.String()))
+			if err != nil {
+				return PageID(INVALID_PAGE), err
+			}
+			_, err = d.fd.WriteAt(d.data, 0)
+			if err != nil {
+				return PageID(INVALID_PAGE), err
+			}
+			d.data = []byte{}
+			d.hasSpilled = &struct{}{}
+		}
+	} else {
+		if d.numPages >= d.onDiskPages {
+			// grow the file by a chunk - 512 pages
+			d.onDiskPages += 512
+			var err error
+			size := int64(d.onDiskPages * PAGE_SIZE)
+			_, err = d.fd.WriteAt([]byte{0}, size-1)
+			if err != nil {
+				return PageID(INVALID_PAGE), err
+			}
+		}
+	}
+
+	return pageID, nil
+}
+
+// DeallocatePage removes page from disk
+func (d *InMemDiskSpillingDiskManager) DeallocatePage(pageID PageID) error {
+	// nothing to do right now
+	return nil
+}
+
+func (d *InMemDiskSpillingDiskManager) FileSize() int64 {
+	return int64(len(d.data))
+}
+
+func (d *InMemDiskSpillingDiskManager) Close() {
+	// close and delete the file if we spilled
+	if d.fd != nil {
+		_ = d.fd.Close()
+		os.Remove(d.fd.Name())
+	}
+}

bufferpool/page.go

@@ -0,0 +1,371 @@
+package bufferpool
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+)
+
+const PAGE_SIZE int = 8192
+
+const INVALID_PAGE int = -1
+
+const PAGE_TYPE_BTREE_INTERNAL = 10
+const PAGE_TYPE_BTREE_LEAF = 11
+const PAGE_TYPE_HASH_TABLE = 12
+
+// PAGE
+// page size 8192 bytes
+// byte aligned, big endian
+
+// |====================================================|
+// | offset | length        |                           |
+// |----------------------------------------------------|
+// | header                                             |
+// |====================================================|
+// | 0      | 4             |  pageNumber (int32)       |
+// | 4      | 2             |  pageType (int16)         |
+// | 6      | 2             |  slotCount (int16)        |
+// | 8      | 2             |  localDepth (int16)       |
+// | 10     | 2             |  freeSpaceOffset (int16)  |
+// | 12     | 4             |  prevPointer (int32)      |
+// | 16     | 4             |  nextPointer (int32)      |
+// |====================================================|
+// | <start of slot array 1..slotCount>                 |
+// |----------------------------------------------------|
+// | 20     | slotcount     |  slot entry is 2 int16    |
+// |        |  * slotwidth  |   values (payloadOffset,  |
+// |        |  * #slots     |   payloadLength)          |
+// |----------------------------------------------------|
+// | <free space>                                       |
+// |----------------------------------------------------|
+// | <payload starting at freeSpaceOffset>              |
+// |  payload entries are keylength (int16), key bytes, |
+// |  payload length (int32), payload bytes             |
+// |====================================================|
+
+const PAGE_NUMBER_OFFSET = 0        // offset 0, length 4, end 4
+const PAGE_TYPE_OFFSET = 4          // offset 4, length 2, end 6
+const PAGE_SLOT_COUNT_OFFSET = 6    // offset 6, length 2, end 8
+const PAGE_LOCAL_DEPTH_OFFSET = 8   // offset 8, length 2, end 10
+const PAGE_FREE_SPACE_OFFSET = 10   // offset 10, length 2, end 12
+const PAGE_PREV_POINTER_OFFSET = 12 // offset 12, length 4, end 16
+const PAGE_NEXT_POINTER_OFFSET = 16 // offset 16, length 4, end 20
+const PAGE_SLOTS_START_OFFSET = 20  // offset 20
+
+// page slots
+//
+//	key offset int16    	//offset 0, length 2, end 2
+//	value offset int16	    //offset 2, length 2, end 4
+const PAGE_SLOT_LENGTH = 4
+
+// Page represents a page on disk
+type Page struct {
+	id       PageID
+	pinCount int
+	isDirty  bool
+	data     [PAGE_SIZE]byte
+}
+
+type PageSlot struct {
+	KeyOffset   int16
+	ValueOffset int16
+}
+
+func (s *PageSlot) KeyBytes(page *Page) []byte {
+	offset := s.KeyOffset
+	keyLen := int16(binary.BigEndian.Uint16(page.data[offset:]))
+	offset += 2
+	result := make([]byte, keyLen)
+	copy(result, page.data[offset:offset+keyLen])
+	return result
+}
+
+func (s *PageSlot) KeyAsInt(page *Page) int32 {
+	return int32(binary.BigEndian.Uint32(page.data[s.KeyOffset+2:]))
+}
+
+func (s *PageSlot) ValueBytes(page *Page) []byte {
+	offset := s.ValueOffset
+	valueLen := int32(binary.BigEndian.Uint32(page.data[offset:]))
+	offset += 4
+	result := make([]byte, valueLen)
+	copy(result, page.data[offset:int32(offset)+valueLen])
+	return result
+}
+
+func (s *PageSlot) ValueAsPagePointer(page *Page) int32 {
+	return int32(binary.BigEndian.Uint32(page.data[s.ValueOffset+4:]))
+}
+
+type PageChunk struct {
+	KeyLength int16
+	KeyBytes  []byte
+	// TODO(pok) ValueBytes can be up to int32 long
+	// this requires an overflow page mechanism, that is not implemented
+	// yet, so be aware of this when storing stuff...
+	ValueLength int32
+	ValueBytes  []byte
+}
+
+func (pc *PageChunk) Length() int {
+	return 2 + len(pc.KeyBytes) + 4 + len(pc.ValueBytes)
+}
+
+func (pc *PageChunk) ComputeKeyOffset(pageOffset int) int {
+	return pageOffset
+}
+
+func (pc *PageChunk) ComputeValueOffset(pageOffset int) int {
+	return pageOffset + 2 + len(pc.KeyBytes)
+}
+
+func (p *Page) WritePageNumber(pageNumber int32) {
+	p.id = PageID(pageNumber)
+	binary.BigEndian.PutUint32(p.data[PAGE_NUMBER_OFFSET:], uint32(pageNumber))
+	p.isDirty = true
+}
+
+func (p *Page) ReadPageNumber() int {
+	return int(binary.BigEndian.Uint32(p.data[PAGE_NUMBER_OFFSET:]))
+}
+
+func (p *Page) WritePageType(pageType int16) {
+	binary.BigEndian.PutUint16(p.data[PAGE_TYPE_OFFSET:], uint16(pageType))
+	p.isDirty = true
+}
+
+func (p *Page) ReadPageType() int16 {
+	return int16(binary.BigEndian.Uint16(p.data[PAGE_TYPE_OFFSET:]))
+}
+
+func (p *Page) WriteSlotCount(slotCount int16) {
+	binary.BigEndian.PutUint16(p.data[PAGE_SLOT_COUNT_OFFSET:], uint16(slotCount))
+	p.isDirty = true
+}
+
+func (p *Page) ReadSlotCount() int16 {
+	return int16(binary.BigEndian.Uint16(p.data[PAGE_SLOT_COUNT_OFFSET:]))
+}
+
+func (p *Page) WriteLocalDepth(localDepth int16) {
+	binary.BigEndian.PutUint16(p.data[PAGE_LOCAL_DEPTH_OFFSET:], uint16(localDepth))
+	p.isDirty = true
+}
+
+func (p *Page) ReadLocalDepth() int16 {
+	return int16(binary.BigEndian.Uint16(p.data[PAGE_LOCAL_DEPTH_OFFSET:]))
+}
+
+func (p *Page) ReadSlot(slot int16) PageSlot {
+	offset := PAGE_SLOTS_START_OFFSET + PAGE_SLOT_LENGTH*slot
+	keyOffset := int16(binary.BigEndian.Uint16(p.data[offset:]))
+	offset += 2
+	valueOffset := int16(binary.BigEndian.Uint16(p.data[offset:]))
+	return PageSlot{
+		KeyOffset:   keyOffset,
+		ValueOffset: valueOffset,
+	}
+}
+
+func (p *Page) WriteSlot(slot int16, value PageSlot) {
+	offset := PAGE_SLOTS_START_OFFSET + PAGE_SLOT_LENGTH*slot
+	binary.BigEndian.PutUint16(p.data[offset:], uint16(value.KeyOffset))
+	offset += 2
+	binary.BigEndian.PutUint16(p.data[offset:], uint16(value.ValueOffset))
+}
+
+func (p *Page) WriteFreeSpaceOffset(offset int16) {
+	binary.BigEndian.PutUint16(p.data[PAGE_FREE_SPACE_OFFSET:], uint16(offset))
+	p.isDirty = true
+}
+
+func (p *Page) ReadFreeSpaceOffset() int16 {
+	return int16(binary.BigEndian.Uint16(p.data[PAGE_FREE_SPACE_OFFSET:]))
+}
+
+func (p *Page) WritePrevPointer(prevPointer int32) {
+	binary.BigEndian.PutUint32(p.data[PAGE_PREV_POINTER_OFFSET:], uint32(prevPointer))
+	p.isDirty = true
+}
+
+func (p *Page) ReadPrevPointer() int {
+	return int(binary.BigEndian.Uint32(p.data[PAGE_PREV_POINTER_OFFSET:]))
+}
+
+func (p *Page) WriteNextPointer(nextPointer int32) {
+	binary.BigEndian.PutUint32(p.data[PAGE_NEXT_POINTER_OFFSET:], uint32(nextPointer))
+	p.isDirty = true
+}
+
+func (p *Page) ReadNextPointer() int {
+	return int(binary.BigEndian.Uint32(p.data[PAGE_NEXT_POINTER_OFFSET:]))
+}
+
+func (p *Page) WriteChunk(offset int16, chunk PageChunk) {
+	binary.BigEndian.PutUint16(p.data[offset:], uint16(chunk.KeyLength))
+	offset += 2
+	copy(p.data[offset:], chunk.KeyBytes)
+	offset += int16(len(chunk.KeyBytes))
+	binary.BigEndian.PutUint32(p.data[offset:], uint32(chunk.ValueLength))
+	offset += 4
+	copy(p.data[offset:], chunk.ValueBytes)
+	p.isDirty = true
+}
+
+func (p *Page) ReadChunk(offset int16) PageChunk {
+	keyLen := int16(binary.BigEndian.Uint16(p.data[offset:]))
+	offset += 2
+	keyBytes := make([]byte, keyLen)
+	copy(keyBytes, p.data[offset:offset+keyLen])
+	offset += keyLen
+	valueLen := int32(binary.BigEndian.Uint32(p.data[offset:]))
+	offset += 4
+	valueBytes := make([]byte, valueLen)
+	copy(valueBytes, p.data[offset:int32(offset)+valueLen])
+	return PageChunk{
+		KeyLength:   keyLen,
+		KeyBytes:    keyBytes,
+		ValueLength: valueLen,
+		ValueBytes:  valueBytes,
+	}
+}
+
+func (p *Page) FreeSpace() int16 {
+	freeSpaceOffset := p.ReadFreeSpaceOffset()
+	freespace := freeSpaceOffset - (p.ReadSlotCount()*PAGE_SLOT_LENGTH + PAGE_SLOT_LENGTH + PAGE_SLOTS_START_OFFSET)
+	return freespace
+}
+
+func (p *Page) WriteKeyValueInSlot(slotNumber int16, key []byte, value []byte) error {
+	freeSpaceOffset := p.ReadFreeSpaceOffset()
+
+	// build a chunk
+	chunk := PageChunk{
+		KeyLength:   int16(len(key)),
+		KeyBytes:    key,
+		ValueLength: int32(len(value)),
+		ValueBytes:  value,
+	}
+
+	// compute the new free space offset
+	freeSpaceOffset -= int16(chunk.Length())
+
+	// check we won't blow free space on page
+	slotCount := p.ReadSlotCount()
+	slotEndOffset := slotCount*PAGE_SLOT_LENGTH + PAGE_SLOT_LENGTH + PAGE_SLOTS_START_OFFSET
+
+	// DEBUG!!
+	//fmt.Printf("freeSpaceOffset: %d, slotCount: %d, slotCount*4 + 4 + 20: %d, freeSpace: %d\n", freeSpaceOffset, slotCount, slotEndOffset, freeSpaceOffset-slotEndOffset)
+
+	if freeSpaceOffset-slotEndOffset <= 0 {
+		return errors.New("page is full")
+	}
+
+	keyOffset := chunk.ComputeKeyOffset(int(freeSpaceOffset))
+	valueOffset := chunk.ComputeValueOffset(int(freeSpaceOffset))
+
+	p.WriteChunk(freeSpaceOffset, chunk)
+
+	// update the free space offset
+	p.WriteFreeSpaceOffset(int16(freeSpaceOffset))
+
+	// make a slot
+	slot := PageSlot{
+		KeyOffset:   int16(keyOffset),
+		ValueOffset: int16(valueOffset),
+	}
+	// write the slot
+	p.WriteSlot(slotNumber, slot)
+
+	return nil
+}
+
+func (p *Page) WritePage(page *Page) {
+	// copy everything but pageNumber & pageType
+	offset := PAGE_SLOT_COUNT_OFFSET
+	copy(page.data[offset:], p.data[offset:offset+PAGE_SIZE-offset])
+}
+
+func (p *Page) PinCount() int {
+	return p.pinCount
+}
+
+func (p *Page) ID() PageID {
+	return p.id
+}
+
+func (p *Page) DecPinCount() {
+	if p.pinCount > 0 {
+		p.pinCount--
+	}
+}
+
+type PageSlotIterator struct {
+	page      *Page
+	slotCount int16
+	cursor    int16
+}
+
+func NewPageSlotIterator(page *Page, fromSlot int16) *PageSlotIterator {
+	i := &PageSlotIterator{
+		page:      page,
+		slotCount: page.ReadSlotCount(),
+		cursor:    fromSlot,
+	}
+	return i
+}
+
+func (i *PageSlotIterator) Next() *PageSlot {
+	if i.cursor < i.slotCount {
+		s := i.page.ReadSlot(i.cursor)
+		i.cursor++
+		return &s
+	}
+	return nil
+}
+
+func (i *PageSlotIterator) Cursor() int16 {
+	return i.cursor
+}
+
+func (pg *Page) Dump(label string) {
+	indent := 0
+	if len(label) > 0 {
+		fmt.Printf("%s%s:\n", fmt.Sprintf("%*s", indent, ""), label)
+		indent += 4
+	}
+	pageType := pg.ReadPageType()
+	fmt.Printf("%sPAGE(%d) pageType: %d slotCount: %d, prevPtr: %d, nextPtr: %d\n", fmt.Sprintf("%*s", indent, ""), pg.ID(), pageType, pg.ReadSlotCount(), pg.ReadPrevPointer(), pg.ReadNextPointer())
+	fmt.Printf("%sKEYS: -->\n", fmt.Sprintf("%*s", indent, ""))
+	indent += 4
+
+	// get the keys off the page
+	keys := make([]int, 0)
+	pointers := make([]int, 0)
+	iter := NewPageSlotIterator(pg, 0)
+	for {
+		ps := iter.Next()
+		if ps == nil {
+			break
+		}
+		keys = append(keys, int(ps.KeyAsInt(pg)))
+		if pageType == /*nodeTypeInternal*/ 10 {
+			pointers = append(pointers, int(ps.ValueAsPagePointer(pg)))
+		}
+	}
+
+	if pageType == /*nodeTypeLeaf*/ 11 {
+		for _, key := range keys {
+			fmt.Printf("%s(%d)\n", fmt.Sprintf("%*s", indent, ""), key)
+		}
+	} else {
+		for idx, key := range keys {
+			ptr := pointers[idx]
+			fmt.Printf("%s(%d, %d)\n", fmt.Sprintf("%*s", indent, ""), key, ptr)
+		}
+		ptr := pg.ReadNextPointer()
+		fmt.Printf("%s(-->, %d)\n", fmt.Sprintf("%*s", indent, ""), ptr)
+	}
+
+}

dax/queryer/orchestrator.go

@@ -2910,13 +2910,13 @@ func (o *orchestrator) howToTranslate(ctx context.Context, idx *featurebase.Inde
 	// First get the index and field the row specifies (if any).
 	rowIdx = idx
 	if row.Index != "" && row.Index != idx.Name {
-		rowIdx, err = o.schemaIndexInfo(ctx, dax.StringTableKeyer(row.Index))
+		rowIdx, err = o.schemaIndexInfo(ctx, dax.TableKey(row.Index))
 		if err != nil {
 			return nil, nil, 0, errors.Wrapf(err, "got a row with unknown index: %s", row.Index)
 		}
 	}
 	if row.Field != "" {
-		rowField, err = o.schemaFieldInfo(ctx, dax.StringTableKeyer(row.Index), row.Field)
+		rowField, err = o.schemaFieldInfo(ctx, dax.TableKey(row.Index), row.Field)
 		if err != nil {
 			return nil, nil, 0, errors.Wrapf(err, "got a row with unknown index/field %s/%s", idx.Name, row.Field)
 		}
@@ -3011,7 +3011,7 @@ func (o *orchestrator) translateResult(ctx context.Context, qtbl *dax.QualifiedT
 			}
 			return other, nil
 		case byRowField:
-			keys, err := o.trans.TranslateFieldListIDs(ctx, rowIdx.Name, rowField.Name, result.Columns())
+			keys, err := o.trans.TranslateFieldListIDs(ctx, result.Index, rowField.Name, result.Columns())
 			if err != nil {
 				return nil, errors.Wrap(err, "translating Row to field keys")
 			}
@@ -3534,6 +3534,12 @@ func (o *orchestrator) schemaFieldInfo(ctx context.Context, tableKeyer dax.Table
 		if err != nil {
 			return nil, errors.Wrapf(err, "getting table by name: %s", v)
 		}
+	case dax.TableKey:
+		qtid := v.QualifiedTableID()
+		tbl, err = o.schema.TableByID(ctx, qtid.ID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "getting table by ID from TableKey: %s", v)
+		}
 	default:
 		return nil, errors.Errorf("unsupport table keyer type in schemaFieldInfo: %T", tableKeyer)
 	}
@@ -3561,6 +3567,12 @@ func (o *orchestrator) schemaIndexInfo(ctx context.Context, tableKeyer dax.Table
 		if err != nil {
 			return nil, errors.Wrapf(err, "getting table by id: %s", v.ID)
 		}
+	case dax.TableKey:
+		qtid := v.QualifiedTableID()
+		tbl, err = o.schema.TableByID(ctx, qtid.ID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "getting table by ID from TableKey: %s", v)
+		}
 	case dax.StringTableKeyer:
 		tbl, err = o.schema.TableByName(ctx, dax.TableName(v))
 		if err != nil {

dax/table.go

@@ -129,6 +129,8 @@ func (s StringTableKeyer) Key() TableKey {
 // TableKey as the value for index.Name.
 type TableKey string
 
+func (t TableKey) Key() TableKey { return t }
+
 // QualifiedTableID returns the QualifiedTableID based on the key. If TableKey
 // can't be parsed into a valid (i.e. complete) QualifiedTableID, then blank
 // values are used where necessary.

extendiblehash/extendiblehash.go

@@ -0,0 +1,266 @@
+package extendiblehash
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/molecula/featurebase/v3/bufferpool"
+)
+
+// ExtendibleHashTable is an extendible hash table implementation backed by a buffer
+// pool
+type ExtendibleHashTable struct {
+	directory   []bufferpool.PageID
+	globalDepth uint
+	keysPerPage int
+	bufferPool  *bufferpool.BufferPool
+}
+
+// NewExtendibleHashTable creates a new ExtendibleHashTable
+func NewExtendibleHashTable(keyLength int, valueLength int, bufferPool *bufferpool.BufferPool) (*ExtendibleHashTable, error) {
+	bytesPerKV := keyLength + valueLength + bufferpool.PAGE_SLOT_LENGTH
+	keysPerPage := (bufferpool.PAGE_SIZE - bufferpool.PAGE_SLOTS_START_OFFSET) / bytesPerKV
+
+	//create the root page
+	page, err := bufferPool.NewPage()
+	if err != nil {
+		return nil, err
+	}
+	page.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+	bufferPool.FlushPage(page.ID())
+
+	return &ExtendibleHashTable{
+		globalDepth: 0,
+		directory:   make([]bufferpool.PageID, 1),
+		keysPerPage: keysPerPage,
+		bufferPool:  bufferPool,
+	}, nil
+}
+
+// Get gets a key from the hash table. It returns the value, a bool set to true if the key is
+// found (false if the key is not found) or an error.
+func (e *ExtendibleHashTable) Get(key []byte) ([]byte, bool, error) {
+	pageId, err := e.getPageID(key)
+	if err != nil {
+		return []byte{}, false, err
+	}
+
+	page, err := e.bufferPool.FetchPage(pageId)
+	if err != nil {
+		return []byte{}, false, err
+	}
+	defer e.bufferPool.UnpinPage(page.ID())
+
+	index, found := e.findKey(page, key)
+	if found {
+		slot := page.ReadSlot(int16(index))
+		return slot.ValueBytes(page), true, nil
+	}
+	return []byte{}, false, nil
+}
+
+// Put puts a key/value pair into the hash table. It returns an error if one occurs.
+func (e *ExtendibleHashTable) Put(key, value []byte) error {
+	pageID, err := e.getPageID(key)
+	if err != nil {
+		return err
+	}
+	page, err := e.bufferPool.FetchPage(pageID)
+	if err != nil {
+		return err
+	}
+	defer e.bufferPool.UnpinPage(page.ID())
+
+	full := int(page.ReadSlotCount()) >= e.keysPerPage
+	err = e.putKeyValue(page, key, value)
+	if err != nil {
+		return err
+	}
+
+	if full {
+		err = e.splitOnKey(page, key)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Close cleans up the hash table after its use.
+func (e *ExtendibleHashTable) Close() {
+	e.bufferPool.Close()
+}
+
+func (e *ExtendibleHashTable) hashFunction(k Hashable) int {
+	hashResult := k.Hash() & ((1 << e.globalDepth) - 1)
+	return int(hashResult)
+}
+
+func (e *ExtendibleHashTable) getPageID(key []byte) (bufferpool.PageID, error) {
+	hash := e.hashFunction(Key(key))
+	if hash > len(e.directory)-1 {
+		return 0, fmt.Errorf("hash (%d) out of the directory array bounds (%d)", hash, len(e.directory))
+	}
+	id := e.directory[hash]
+	return bufferpool.PageID(id), nil
+}
+
+func (e *ExtendibleHashTable) findKey(page *bufferpool.Page, key []byte) (int, bool) {
+	minIndex := 0
+	onePastMaxIndex := int(page.ReadSlotCount())
+
+	for onePastMaxIndex != minIndex {
+		index := (minIndex + onePastMaxIndex) / 2
+		s := page.ReadSlot(int16(index))
+		keyAtIndex := s.KeyBytes(page)
+
+		if bytes.Equal(keyAtIndex, key) {
+			return index, true
+		}
+		if bytes.Compare(key, keyAtIndex) < 0 {
+			onePastMaxIndex = index
+		} else {
+			minIndex = index + 1
+		}
+	}
+	return minIndex, false
+}
+
+func (e *ExtendibleHashTable) splitOnKey(page *bufferpool.Page, key []byte) error {
+	if uint(page.ReadLocalDepth()) == e.globalDepth {
+
+		e.directory = append(e.directory, e.directory...)
+		e.globalDepth++
+	}
+
+	// scratch page for left
+	p0 := e.bufferPool.ScratchPage()
+	p0.WritePageNumber(int32(page.ID()))
+	p0.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+
+	// allocate new page for split
+	p1, err := e.bufferPool.NewPage()
+	if err != nil {
+		return err
+	}
+	defer e.bufferPool.UnpinPage(p1.ID())
+	p1.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+
+	// update local depths
+	newLocalDepth := page.ReadLocalDepth() + 1
+	p0.WriteLocalDepth(newLocalDepth)
+	p1.WriteLocalDepth(newLocalDepth)
+
+	ld := page.ReadLocalDepth()
+	hiBit := uint64(1 << ld)
+
+	it := bufferpool.NewPageSlotIterator(page, 0)
+	for {
+		slot := it.Next()
+		if slot == nil {
+			break
+		}
+		keyBytes := slot.KeyBytes(page)
+		k := string(keyBytes)
+		h := Key(k).Hash()
+
+		if h&hiBit > 0 {
+			sc := p1.ReadSlotCount()
+			p1.WriteKeyValueInSlot(sc, keyBytes, slot.ValueBytes(page))
+			// update the slot count
+			p1.WriteSlotCount(int16(sc + 1))
+
+		} else {
+			sc := p0.ReadSlotCount()
+			p0.WriteKeyValueInSlot(sc, keyBytes, slot.ValueBytes(page))
+			// update the slot count
+			p0.WriteSlotCount(int16(sc + 1))
+		}
+	}
+	for j := Key(key).Hash() & (hiBit - 1); j < uint64(len(e.directory)); j += hiBit {
+		if j&hiBit > 0 {
+			e.directory[j] = p1.ID()
+		} else {
+			e.directory[j] = p0.ID()
+		}
+	}
+
+	// copy p1 back into page
+	p0.WritePage(page)
+
+	return nil
+}
+
+func (e *ExtendibleHashTable) cleanPage(page *bufferpool.Page) error {
+	scratch := e.bufferPool.ScratchPage()
+	// copy page number
+	scratch.WritePageNumber(int32(page.ID()))
+	// set the page type
+	scratch.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+	// copy local depth
+	scratch.WriteLocalDepth(page.ReadLocalDepth())
+
+	// copy slots from page to scratch
+	si := bufferpool.NewPageSlotIterator(page, 0)
+	for {
+		slot := si.Next()
+		if slot == nil {
+			break
+		}
+		scratch.WriteKeyValueInSlot(si.Cursor(), slot.KeyBytes(page), slot.ValueBytes(page))
+	}
+
+	// update the slot count
+	scratch.WriteSlotCount(page.ReadSlotCount())
+
+	// write scratch back to page
+	scratch.WritePage(page)
+	return nil
+}
+
+func (e *ExtendibleHashTable) keyValueWillFit(page *bufferpool.Page, key, value []byte) bool {
+	// will this k/v fit on the page?
+	slotLen := 4                          // we need 2 len words for the slot
+	chunkLen := 6 + len(key) + len(value) // int16 len + int32 len + len of respective []byte
+	fs := page.FreeSpace()
+	return fs > (int16(slotLen) + int16(chunkLen))
+}
+
+func (e *ExtendibleHashTable) putKeyValue(page *bufferpool.Page, key, value []byte) error {
+
+	if !e.keyValueWillFit(page, key, value) {
+		// try to garbage collect the page first
+		err := e.cleanPage(page)
+		if err != nil {
+			return err
+		}
+	}
+
+	//find the key
+	newIndex, found := e.findKey(page, []byte(key))
+	// get the slot count
+	slotCount := int(page.ReadSlotCount())
+	if found {
+		// we found the key, so we will update the value
+		err := page.WriteKeyValueInSlot(int16(newIndex), []byte(key), []byte(value))
+		if err != nil {
+			return err
+		}
+	} else {
+		// TODO(pok) should check in WriteSlot() to see if we are out space too...
+
+		// TODO(pok) we should move all the slots in one fell swoop, because,... performance
+		// move all the slots after where we are going to insert
+		for j := slotCount; j > newIndex; j-- {
+			sl := page.ReadSlot(int16(j - 1))
+			page.WriteSlot(int16(j), sl)
+		}
+		err := page.WriteKeyValueInSlot(int16(newIndex), []byte(key), []byte(value))
+		if err != nil {
+			return err
+		}
+		// update the slot count
+		page.WriteSlotCount(int16(slotCount + 1))
+	}
+	return nil
+}

extendiblehash/extendiblehash_test.go

@@ -0,0 +1,327 @@
+package extendiblehash
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/molecula/featurebase/v3/bufferpool"
+	"github.com/stretchr/testify/assert"
+)
+
+func makeDirectory() (*ExtendibleHashTable, error) {
+	diskManager := bufferpool.NewInMemDiskSpillingDiskManager(128)
+	bufferPool := bufferpool.NewBufferPool(128, diskManager)
+
+	keySize := 12
+	valueSize := 20
+
+	return NewExtendibleHashTable(keySize, valueSize, bufferPool)
+}
+
+func TestHashTable_ExtendibleHash(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.globalDepth = 4
+
+	key := "321"  //   0011
+	key2 := "123" //  1011
+
+	result := d.hashFunction(Key(key))
+	result2 := d.hashFunction(Key(key2))
+
+	assert.Equal(t, 7, result)
+	assert.Equal(t, 6, result2)
+}
+
+func TestHashTable_GetPage(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.globalDepth = 4
+	d.directory = make([]bufferpool.PageID, 16)
+
+	key := "478"
+	d.directory[14] = 2
+
+	pageID, err := d.getPageID([]byte(key))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, 2, int(pageID))
+}
+
+func TestHashTable_GetPage_ShouldReturnError_WhenOffsetIsNotLimitedToDataSize(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.globalDepth = 4
+	key := "478"
+
+	_, err = d.getPageID([]byte(key))
+	assert.Error(t, err)
+}
+
+func TestHashTable_GetPage_ShouldReturnError_WhenPageIDIsOutOfTheTable(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.directory = make([]bufferpool.PageID, 0)
+	key := "123"
+
+	_, err = d.getPageID([]byte(key))
+	assert.Error(t, err)
+}
+
+func TestHashTable_Get(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.globalDepth = 4
+	d.directory = make([]bufferpool.PageID, 16)
+
+	d.directory[14] = 2
+
+	// force there to be two pages
+	page, err := d.bufferPool.NewPage() //1
+	if err != nil {
+		t.Fatal(err)
+	}
+	page.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+	d.bufferPool.FlushPage(page.ID())
+
+	page, err = d.bufferPool.NewPage() //2
+	if err != nil {
+		t.Fatal(err)
+	}
+	page.WritePageType(bufferpool.PAGE_TYPE_HASH_TABLE)
+	d.bufferPool.FlushPage(page.ID())
+
+	// now do the test
+	page, err = d.bufferPool.FetchPage(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(page.ID())
+
+	key := "478"
+	value := "Hi"
+
+	page.WriteKeyValueInSlot(0, []byte(key), []byte(value))
+	page.WriteSlotCount(int16(1))
+
+	result, _, err := d.Get([]byte(key))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, "Hi", string(result))
+}
+
+func TestHashTable_Get_ShouldHandleError(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+	key := "123"
+
+	result, found, err := d.Get([]byte(key))
+
+	assert.Equal(t, err, nil)
+	assert.Equal(t, []byte{}, result)
+	assert.Equal(t, false, found)
+}
+
+func TestHashTable_Put(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	page, err := d.bufferPool.FetchPage(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(page.ID())
+	err = addToPage(page, 5)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d.Put([]byte("123"), []byte("Yolo !"))
+
+	value, found, err := d.Get([]byte("123"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, "Yolo !", string(value))
+	assert.Equal(t, true, found)
+}
+
+func TestHashTable_Put_ShouldIncreaseSize_WhenTableIsFull(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	page, err := d.bufferPool.FetchPage(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(page.ID())
+	err = addToPage(page, 227) // keys per page with key 12, value 20
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d.Put([]byte("123"), []byte("Yolo !"))
+
+	value, _, err := d.Get([]byte("123"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, "Yolo !", string(value))
+	assert.Equal(t, 2, len(d.directory))
+	assert.Equal(t, uint(1), d.globalDepth)
+}
+
+func TestHashTable_PutShouldIncrementLD_WhenPageIsFull(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	page, err := d.bufferPool.FetchPage(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(page.ID())
+	err = addToPage(page, 227) // keys per page with key 12, value 20
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d.Put([]byte("12345678"), []byte("Yolo !"))
+
+	assert.Equal(t, int64(8192*2), d.bufferPool.OnDiskSize())
+	assert.Equal(t, 1, int(d.globalDepth))
+
+	p0, err := d.bufferPool.FetchPage(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(p0.ID())
+	assert.Equal(t, int16(1), p0.ReadLocalDepth())
+
+	p1, err := d.bufferPool.FetchPage(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.bufferPool.UnpinPage(p1.ID())
+	assert.Equal(t, int16(1), p1.ReadLocalDepth())
+}
+
+func TestHashTable_Put_INT(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 4000; i++ {
+		err = d.Put([]byte("key"+strconv.Itoa(i)), []byte("Yolo !"))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	assert.Equal(t, []bufferpool.PageID{0, 1, 2, 3, 4, 7, 6, 5, 13, 14, 12, 9, 8, 15, 10, 11, 28, 24, 21, 18, 4, 19, 29, 20, 27, 22, 25, 23, 17, 15, 16, 26}, d.directory)
+}
+
+func TestHashTable_Put_SameKey_ALotOfTime(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 10000; i++ {
+		d.Put([]byte("key"), []byte("Yolo ! "+strconv.Itoa(i)))
+	}
+
+	value, _, err := d.Get([]byte("key"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, "Yolo ! 9999", string(value))
+	assert.Equal(t, 1, len(d.directory))
+	assert.Equal(t, int64(8192), d.bufferPool.OnDiskSize())
+}
+
+func TestHashTable_Put_Many_Keys(t *testing.T) {
+	d, err := makeDirectory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 1000000; i++ {
+		err = d.Put([]byte("key"+strconv.Itoa(i)), []byte("Yolo ! "+strconv.Itoa(i)))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	value, _, err := d.Get([]byte("key99756"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, "Yolo ! 99756", string(value))
+	assert.Equal(t, 8192, len(d.directory))
+	assert.Equal(t, uint(13), d.globalDepth)
+	d.Close()
+}
+
+func BenchmarkHashTable_Put_Many_Keys(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+
+		d, err := makeDirectory()
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		for i := 0; i < 1000000; i++ {
+			err = d.Put([]byte("key"+strconv.Itoa(i)), []byte("Yolo ! "+strconv.Itoa(i)))
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+
+		value, _, err := d.Get([]byte("key99756"))
+		if err != nil {
+			b.Fatal(err)
+		}
+		assert.Equal(b, "Yolo ! 99756", string(value))
+		assert.Equal(b, 8192, len(d.directory))
+		assert.Equal(b, uint(13), d.globalDepth)
+		d.Close()
+
+	}
+}
+
+func addToPage(page *bufferpool.Page, numberOfRecords int) error {
+	for i := 0; i < numberOfRecords; i++ {
+		//fmt.Printf("writing record %d\n", i+1)
+		itoa := strconv.Itoa(i)
+		err := page.WriteKeyValueInSlot(int16(i), []byte("key"+itoa), []byte("value foo bar"))
+		if err != nil {
+			return err
+		}
+		page.WriteSlotCount(int16(page.ReadSlotCount() + 1))
+	}
+	return nil
+}

extendiblehash/key.go

@@ -0,0 +1,24 @@
+package extendiblehash
+
+import (
+	"github.com/zeebo/xxh3"
+)
+
+type Hashable interface {
+	Hash() uint64
+}
+
+// use the same seed all the time - this is not for crypto
+var protoSeed uint64 = 20041973
+
+var hasher = xxh3.NewSeed(protoSeed)
+
+type Key []byte
+
+// BEWARE - not concurrent!!
+func (k Key) Hash() uint64 {
+	hasher.Reset()
+	hasher.Write(k)
+	hash := hasher.Sum64()
+	return hash
+}

go.mod

@@ -162,7 +162,7 @@ require (
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20201229170055-e5319fda7802 // indirect
 	github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
-	github.com/zeebo/xxh3 v1.0.2 // indirect
+	github.com/zeebo/xxh3 v1.0.2
 	go.etcd.io/etcd/client/v2 v2.305.5 // indirect
 	go.etcd.io/etcd/pkg/v3 v3.5.5 // indirect
 	go.etcd.io/etcd/raft/v3 v3.5.5 // indirect

sql3/planner/compileselect.go

@@ -46,11 +46,6 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
 	}
 	var err error
 
-	// handle distinct
-	if stmt.Distinct.IsValid() {
-		query.AddWarning("DISTINCT not yet implemented")
-	}
-
 	// handle the where clause
 	where, err := p.compileExpr(stmt.WhereExpr)
 	if err != nil {
@@ -223,6 +218,11 @@ func (p *ExecutionPlanner) compileSelectStatement(stmt *parser.SelectStatement,
 		compiledOp = NewPlanOpTop(topExpr, compiledOp)
 	}
 
+	// handle distinct
+	if stmt.Distinct.IsValid() {
+		compiledOp = NewPlanOpDistinct(p, compiledOp)
+	}
+
 	// if it is a subquery, don't wrap in a PlanOpQuery
 	if isSubquery {
 		return compiledOp, nil

sql3/planner/expression.go

@@ -173,6 +173,7 @@ func (n *unaryOpPlanExpression) String() string {
 func (n *unaryOpPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["op"] = n.op
 	result["rhs"] = n.rhs.Plan()
@@ -668,6 +669,7 @@ func (n *binOpPlanExpression) String() string {
 func (n *binOpPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["op"] = n.op
 	result["lhs"] = n.lhs.Plan()
@@ -739,6 +741,7 @@ func (n *rangePlanExpression) String() string {
 func (n *rangePlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["lhs"] = n.lhs.Plan()
 	result["rhs"] = n.rhs.Plan()
@@ -954,6 +957,7 @@ func (n *casePlanExpression) String() string {
 func (n *casePlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	if n.baseExpr != nil {
 		result["baseExpr"] = n.baseExpr.Plan()
@@ -1039,6 +1043,7 @@ func (n *caseBlockPlanExpression) String() string {
 func (n *caseBlockPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["condition"] = n.condition.Plan()
 	result["body"] = n.body.Plan()
@@ -1108,6 +1113,7 @@ func (n *subqueryPlanExpression) String() string {
 func (n *subqueryPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["subquery"] = n.op.Plan()
 	return result
@@ -1216,6 +1222,7 @@ func (n *betweenOpPlanExpression) String() string {
 func (n *betweenOpPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["lhs"] = n.lhs.Plan()
 	result["rhs"] = n.rhs.Plan()
@@ -1434,6 +1441,7 @@ func (n *inOpPlanExpression) String() string {
 func (n *inOpPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["lhs"] = n.lhs.Plan()
 	result["rhs"] = n.rhs.Plan()
@@ -1527,6 +1535,7 @@ func (n *callPlanExpression) String() string {
 func (n *callPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["name"] = n.name
 	result["dataType"] = n.Type().TypeDescription()
 	ps := make([]interface{}, 0)
@@ -1583,6 +1592,7 @@ func (n *aliasPlanExpression) String() string {
 func (n *aliasPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["aliasName"] = n.aliasName
 	result["expr"] = n.expr.Plan()
@@ -1676,6 +1686,7 @@ func (n *qualifiedRefPlanExpression) String() string {
 func (n *qualifiedRefPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["tableName"] = n.tableName
 	result["columnName"] = n.columnName
 	result["columnIndex"] = n.columnIndex
@@ -1738,6 +1749,7 @@ func (n *variableRefPlanExpression) String() string {
 func (n *variableRefPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["name"] = n.name
 	result["dataType"] = n.dataType.TypeDescription()
 	return result
@@ -1773,6 +1785,7 @@ func (n *nullLiteralPlanExpression) String() string {
 func (n *nullLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	return result
 }
@@ -1811,6 +1824,7 @@ func (n *intLiteralPlanExpression) String() string {
 func (n *intLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["value"] = n.value
 	return result
@@ -1851,6 +1865,7 @@ func (n *floatLiteralPlanExpression) String() string {
 func (n *floatLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["value"] = n.value
 	return result
@@ -1890,6 +1905,7 @@ func (n *boolLiteralPlanExpression) String() string {
 func (n *boolLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["value"] = n.value
 	return result
@@ -1929,6 +1945,7 @@ func (n *dateLiteralPlanExpression) String() string {
 func (n *dateLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["value"] = n.value
 	return result
@@ -1968,6 +1985,7 @@ func (n *stringLiteralPlanExpression) String() string {
 func (n *stringLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["value"] = n.value
 	return result
@@ -2192,6 +2210,7 @@ func (n *castPlanExpression) String() string {
 func (n *castPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["lhs"] = n.lhs.Plan()
 	return result
@@ -2243,6 +2262,7 @@ func (n *exprListPlanExpression) String() string {
 func (n *exprListPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	ps := make([]interface{}, 0)
 	for _, e := range n.exprs {
 		ps = append(ps, e.Plan())
@@ -2333,6 +2353,7 @@ func (n *exprSetLiteralPlanExpression) String() string {
 func (n *exprSetLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	ps := make([]interface{}, 0)
 	for _, e := range n.members {
 		ps = append(ps, e.Plan())
@@ -2414,6 +2435,7 @@ func (n *exprTupleLiteralPlanExpression) String() string {
 func (n *exprTupleLiteralPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	ps := make([]interface{}, 0)
 	for _, e := range n.members {
 		ps = append(ps, e.Plan())

sql3/planner/expressionagg.go

@@ -126,6 +126,7 @@ func (n *countPlanExpression) String() string {
 func (n *countPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -194,6 +195,7 @@ func (n *countDistinctPlanExpression) String() string {
 func (n *countDistinctPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -327,6 +329,7 @@ func (n *sumPlanExpression) String() string {
 func (n *sumPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -500,6 +503,7 @@ func (n *avgPlanExpression) String() string {
 func (n *avgPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -641,6 +645,7 @@ func (n *minPlanExpression) String() string {
 func (n *minPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -783,6 +788,7 @@ func (n *maxPlanExpression) String() string {
 func (n *maxPlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	return result
@@ -855,6 +861,7 @@ func (n *percentilePlanExpression) String() string {
 func (n *percentilePlanExpression) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_expr"] = fmt.Sprintf("%T", n)
+	result["description"] = n.String()
 	result["dataType"] = n.Type().TypeDescription()
 	result["arg"] = n.arg.Plan()
 	result["ntharg"] = n.nthArg.Plan()

sql3/planner/memoryobj.go

@@ -1,122 +0,0 @@
-// Copyright 2022 Molecula Corp. All rights reserved.
-
-package planner
-
-import (
-	"hash/maphash"
-
-	"github.com/featurebasedb/featurebase/v3/sql3"
-	"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
-)
-
-var prototypeHash maphash.Hash
-
-// ObjectCache is a cache of interface{} values
-type ObjectCache interface {
-	// Put a new value in the cache
-	PutObject(uint64, interface{}) error
-
-	// Get the value with the given key
-	GetObject(uint64) (interface{}, error)
-
-	// Size returns the number of values in the cache
-	Size() int
-}
-
-// RowCache is a cache of rows used during row iteration
-type RowCache interface {
-	Add(row types.Row) error
-
-	// AllRows returns all rows.
-	AllRows() []types.Row
-}
-
-// KeyedRowCache is a cache of keyed rows used during row iteration
-type KeyedRowCache interface {
-	// Put adds row to the cache at the given key.
-	Put(key uint64, row types.Row) error
-
-	// Get returns the rows specified by key.
-	Get(key uint64) (types.Row, error)
-
-	// Size returns the number of rows in the cache.
-	Size() int
-}
-
-// Ensure type implements interface
-var _ KeyedRowCache = (*inMemoryKeyedRowCache)(nil)
-
-// default implementation of KeyedRowCache (in memory)
-type inMemoryKeyedRowCache struct {
-	store map[uint64][]interface{}
-}
-
-func newinMemoryKeyedRowCache() *inMemoryKeyedRowCache {
-	return &inMemoryKeyedRowCache{
-		store: make(map[uint64][]interface{}),
-	}
-}
-
-func (m inMemoryKeyedRowCache) Put(u uint64, i types.Row) error {
-	m.store[u] = i
-	return nil
-}
-
-func (m inMemoryKeyedRowCache) Get(u uint64) (types.Row, error) {
-	return m.store[u], nil
-}
-
-func (m inMemoryKeyedRowCache) Size() int {
-	return len(m.store)
-}
-
-// Ensure type implements interface
-var _ RowCache = (*inMemoryRowCache)(nil)
-
-type inMemoryRowCache struct {
-	rows []types.Row
-}
-
-func newInMemoryRowCache() *inMemoryRowCache {
-	return &inMemoryRowCache{}
-}
-
-func (c *inMemoryRowCache) Add(row types.Row) error {
-	c.rows = append(c.rows, row)
-	return nil
-}
-
-func (c *inMemoryRowCache) AllRows() []types.Row {
-	return c.rows
-}
-
-// Ensure type implements interface
-var _ ObjectCache = (*mapObjectCache)(nil)
-
-// mapObjectCache is a simple in-memory implementation of a cache
-type mapObjectCache struct {
-	cache map[uint64]interface{}
-}
-
-func (m mapObjectCache) PutObject(u uint64, i interface{}) error {
-	m.cache[u] = i
-	return nil
-}
-
-func (m mapObjectCache) GetObject(u uint64) (interface{}, error) {
-	v, ok := m.cache[u]
-	if !ok {
-		return nil, sql3.NewErrCacheKeyNotFound(u)
-	}
-	return v, nil
-}
-
-func (m mapObjectCache) Size() int {
-	return len(m.cache)
-}
-
-func NewMapObjectCache() mapObjectCache {
-	return mapObjectCache{
-		cache: make(map[uint64]interface{}),
-	}
-}

sql3/planner/opbulkinsert.go

@@ -76,11 +76,7 @@ func NewPlanOpBulkInsert(p *ExecutionPlanner, tableName string, options *bulkIns
 func (p *PlanOpBulkInsert) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 
 	options := make(map[string]interface{})

sql3/planner/opcreatetable.go

@@ -41,11 +41,6 @@ func NewPlanOpCreateTable(p *ExecutionPlanner, tableName string, failIfExists bo
 func (p *PlanOpCreateTable) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
 	result["name"] = p.tableName
 	result["failIfExists"] = p.failIfExists
 	return result

sql3/planner/opdistinct.go

@@ -3,39 +3,163 @@
 package planner
 
 import (
+	"bytes"
+	"context"
 	"fmt"
 
 	"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
 )
 
 // PlanOpDistinct plan operator handles DISTINCT
+// DISTINCT returns unique rows from its iterator and does this by
+// creating a hash table and probing new rows against that hash table,
+// if the row has already been seen, it is skipped, it it has not been
+// seen, a 'key' is created from all the values in the row and this is
+// inserted into the hash table.
+// The hash table is implemented using Extendible Hashing and is backed
+// by a buffer pool. The buffer pool is allocated to 128 pages (or 1Mb)
+// and the disk manager used by the buffer pool will use an in-memory
+// implementation up to 128 pages and thereafter spill to disk
 type PlanOpDistinct struct {
 	planner  *ExecutionPlanner
-	source   types.PlanOperator
+	ChildOp  types.PlanOperator
 	warnings []string
 }
 
-func NewPlanOpDistinct(p *ExecutionPlanner, source types.PlanOperator) *PlanOpDistinct {
+func NewPlanOpDistinct(p *ExecutionPlanner, child types.PlanOperator) *PlanOpDistinct {
 	return &PlanOpDistinct{
 		planner:  p,
-		source:   source,
+		ChildOp:  child,
 		warnings: make([]string, 0),
 	}
 }
 
-func (n *PlanOpDistinct) Plan() map[string]interface{} {
+func (p *PlanOpDistinct) Schema() types.Schema {
+	return p.ChildOp.Schema()
+}
+
+func (p *PlanOpDistinct) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
+	i, err := p.ChildOp.Iterator(ctx, row)
+	if err != nil {
+		return nil, err
+	}
+	return newDistinctIterator(p.Schema(), i), nil
+}
+
+func (p *PlanOpDistinct) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
+	if len(children) != 1 {
+		return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
+	}
+	return NewPlanOpDistinct(p.planner, children[0]), nil
+}
+
+func (p *PlanOpDistinct) Children() []types.PlanOperator {
+	return []types.PlanOperator{
+		p.ChildOp,
+	}
+}
+
+func (p *PlanOpDistinct) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
-	result["_op"] = fmt.Sprintf("%T", n)
+	result["_op"] = fmt.Sprintf("%T", p)
+	result["_schema"] = p.Schema().Plan()
+	result["child"] = p.ChildOp.Plan()
 	return result
 }
 
-func (n *PlanOpDistinct) AddWarning(warning string) {
-	n.warnings = append(n.warnings, warning)
+func (p *PlanOpDistinct) String() string {
+	return ""
 }
 
-func (n *PlanOpDistinct) Warnings() []string {
+func (p *PlanOpDistinct) AddWarning(warning string) {
+	p.warnings = append(p.warnings, warning)
+}
+
+func (p *PlanOpDistinct) Warnings() []string {
 	var w []string
-	w = append(w, n.warnings...)
-	w = append(w, n.source.Warnings()...)
+	w = append(w, p.warnings...)
+	w = append(w, p.ChildOp.Warnings()...)
 	return w
 }
+
+type distinctIterator struct {
+	child      types.RowIterator
+	schema     types.Schema
+	hasStarted *struct{}
+	hashTable  *extendiblehash.ExtendibleHashTable
+}
+
+func newDistinctIterator(schema types.Schema, child types.RowIterator) *distinctIterator {
+	return &distinctIterator{
+		schema: schema,
+		child:  child,
+	}
+}
+
+func (i *distinctIterator) rowSeen(ctx context.Context, row types.Row) (bool, error) {
+	keyBytes := generateRowKey(row)
+	_, found, err := i.hashTable.Get(keyBytes)
+	if err != nil {
+		return false, nil
+	}
+	// put the row in the hash table to recored that we've seen it
+	if !found {
+		i.hashTable.Put(keyBytes, []byte{1})
+	}
+	return found, nil
+}
+
+func (i *distinctIterator) Next(ctx context.Context) (types.Row, error) {
+	if i.hasStarted == nil {
+		//create the hashtable
+
+		// ask the diskmanager to spill after 1Mb (128 8K pages)
+		diskManager := bufferpool.NewInMemDiskSpillingDiskManager(128)
+		// use 1Mb (128 8K pages)
+		bufferPool := bufferpool.NewBufferPool(128, diskManager)
+
+		// we're going to use something pretty conservative here - we could have sets, strings, ... all sorts of stuff
+		// and we don't yet support long key overflowing
+		keyLength := 128 // bytes
+
+		valueLength := 1 // we're going to store a 1 (byte) for every key in the table
+
+		ht, err := extendiblehash.NewExtendibleHashTable(keyLength, valueLength, bufferPool)
+		if err != nil {
+			return nil, err
+		}
+		i.hashTable = ht
+		i.hasStarted = &struct{}{}
+	}
+
+	for {
+		row, err := i.child.Next(ctx)
+		if err != nil {
+			// clean up
+			// TODO(pok) - we need to move clean up to higher level, and
+			// implement at the operator level
+			if err == types.ErrNoMoreRows {
+				i.hashTable.Close()
+			}
+			return nil, err
+		}
+		// does row exist in hash table
+		seen, err := i.rowSeen(ctx, row)
+		if err != nil {
+			return nil, err
+		}
+		// if we've seen it before, go to the next row
+		if seen {
+			continue
+		}
+		return row, nil
+	}
+}
+
+func generateRowKey(row types.Row) []byte {
+	var buf bytes.Buffer
+	for _, v := range row {
+		buf.WriteString(fmt.Sprintf("%#v", v))
+	}
+	return buf.Bytes()
+}

sql3/planner/opdroptable.go

@@ -29,11 +29,6 @@ func NewPlanOpDropTable(p *ExecutionPlanner, index *pilosa.IndexInfo) *PlanOpDro
 func (p *PlanOpDropTable) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
 	result["tableName"] = p.index.Name
 	return result
 }

sql3/planner/opfeaturebasecolumns.go

@@ -29,11 +29,7 @@ func NewPlanOpFeatureBaseColumns(tbl *dax.Table) *PlanOpFeatureBaseColumns {
 func (p *PlanOpFeatureBaseColumns) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	return result
 }
 

sql3/planner/opfeaturebasetables.go

@@ -29,11 +29,7 @@ func NewPlanOpFeatureBaseTables(indexInfo []*pilosa.IndexInfo) *PlanOpFeatureBas
 func (p *PlanOpFeatureBaseTables) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	return result
 }
 

sql3/planner/opfilter.go

@@ -56,11 +56,7 @@ func (p *PlanOpFilter) Children() []types.PlanOperator {
 func (p *PlanOpFilter) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["predicate"] = p.Predicate.Plan()
 	result["child"] = p.ChildOp.Plan()
 	return result

sql3/planner/opgroupby.go

@@ -3,9 +3,9 @@
 package planner
 
 import (
+	"bytes"
 	"context"
 	"fmt"
-	"hash/maphash"
 
 	"github.com/featurebasedb/featurebase/v3/errors"
 	"github.com/featurebasedb/featurebase/v3/sql3"
@@ -100,11 +100,7 @@ func (p *PlanOpGroupBy) WithUpdatedExpressions(exprs ...types.PlanExpression) (t
 func (p *PlanOpGroupBy) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
+	result["_schema"] = p.Schema().Plan()
 	result["child"] = p.ChildOp.Plan()
 	ps := make([]interface{}, 0)
 	for _, e := range p.Aggregates {
@@ -193,8 +189,8 @@ type keysAndAggregations struct {
 type groupByGroupingIter struct {
 	aggregates   []types.PlanExpression
 	groupByExprs []types.PlanExpression
-	aggregations ObjectCache
-	keys         []uint64
+	aggregations map[string]*keysAndAggregations
+	keys         []string
 	child        types.RowIterator
 }
 
@@ -208,16 +204,16 @@ func newGroupByGroupingIter(ctx context.Context, aggregates, groupByExprs []type
 
 func (i *groupByGroupingIter) Next(ctx context.Context) (types.Row, error) {
 	if i.aggregations == nil {
-		i.aggregations = NewMapObjectCache()
+		i.aggregations = make(map[string]*keysAndAggregations)
 		if err := i.compute(ctx); err != nil {
 			return nil, err
 		}
 	}
 
 	if len(i.keys) > 0 {
-		buffers, err := i.get(i.keys[0])
-		if err != nil {
-			return nil, err
+		buffers, ok := i.aggregations[i.keys[0]]
+		if !ok {
+			return nil, sql3.NewErrInternalf("unexpected absence of key")
 		}
 
 		i.keys = i.keys[1:]
@@ -245,13 +241,13 @@ func (i *groupByGroupingIter) compute(ctx context.Context) error {
 			return err
 		}
 
-		key, keyValues, err := groupingKeyHash(ctx, i.groupByExprs, row)
+		key, keyValues, err := groupingKey(ctx, i.groupByExprs, row)
 		if err != nil {
 			return err
 		}
 
-		b, err := i.get(key)
-		if errors.Is(err, sql3.ErrCacheKeyNotFound) {
+		b, ok := i.aggregations[key]
+		if !ok {
 			b = &keysAndAggregations{}
 			b.buffers = make([]types.AggregationBuffer, len(i.aggregates))
 			for j, a := range i.aggregates {
@@ -261,9 +257,7 @@ func (i *groupByGroupingIter) compute(ctx context.Context) error {
 				}
 			}
 			b.groupByKeys = keyValues
-			if err := i.aggregations.PutObject(key, b); err != nil {
-				return err
-			}
+			i.aggregations[key] = b
 			i.keys = append(i.keys, key)
 		} else if err != nil {
 			return err
@@ -277,17 +271,6 @@ func (i *groupByGroupingIter) compute(ctx context.Context) error {
 	return nil
 }
 
-func (i *groupByGroupingIter) get(key uint64) (*keysAndAggregations, error) {
-	v, err := i.aggregations.GetObject(key)
-	if err != nil {
-		return nil, err
-	}
-	if v == nil {
-		return nil, nil
-	}
-	return v.(*keysAndAggregations), err
-}
-
 func newAggregationBuffer(expr types.PlanExpression) (types.AggregationBuffer, error) {
 	switch n := expr.(type) {
 	case types.Aggregable:
@@ -318,21 +301,16 @@ func evalBuffers(ctx context.Context, aggregationBuffers *keysAndAggregations) (
 	return row, nil
 }
 
-func groupingKeyHash(ctx context.Context, groupByExprs []types.PlanExpression, row types.Row) (uint64, types.Row, error) {
+func groupingKey(ctx context.Context, groupByExprs []types.PlanExpression, row types.Row) (string, types.Row, error) {
+	var buf bytes.Buffer
 	rowKeys := make([]interface{}, len(groupByExprs))
-	var hash maphash.Hash
-	hash.SetSeed(prototypeHash.Seed())
 	for i, expr := range groupByExprs {
 		v, err := expr.Evaluate(row)
 		if err != nil {
-			return 0, nil, err
-		}
-		_, err = hash.Write(([]byte)(fmt.Sprintf("%#v,", v)))
-		if err != nil {
-			return 0, nil, err
+			return "", nil, err
 		}
+		buf.WriteString(fmt.Sprintf("%#v", v))
 		rowKeys[i] = v
 	}
-	result := hash.Sum64()
-	return result, rowKeys, nil
+	return buf.String(), rowKeys, nil
 }

sql3/planner/ophaving.go

@@ -56,11 +56,7 @@ func (p *PlanOpHaving) Children() []types.PlanOperator {
 func (p *PlanOpHaving) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["predicate"] = p.Predicate.Plan()
 	result["child"] = p.ChildOp.Plan()
 	return result

sql3/planner/opinsert.go

@@ -39,26 +39,14 @@ func NewPlanOpInsert(p *ExecutionPlanner, tableName string, targetColumns []*qua
 func (p *PlanOpInsert) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 	ps := make([]interface{}, 0)
 	for _, e := range p.targetColumns {
 		ps = append(ps, e.Plan())
 	}
 	result["targetColumns"] = ps
-	pps := make([]interface{}, 0)
-	for _, tuple := range p.insertValues {
-		ps := make([]interface{}, 0)
-		for _, e := range tuple {
-			ps = append(ps, e.Plan())
-		}
-		pps = append(pps, ps)
-	}
-	result["insertValues"] = pps
+	result["insertTupleCount"] = len(p.insertValues)
 	return result
 }
 

sql3/planner/opnestedloops.go

@@ -31,11 +31,7 @@ func NewPlanOpNestedLoops(top, bottom types.PlanOperator, condition types.PlanEx
 func (p *PlanOpNestedLoops) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["top"] = p.top.Plan()
 	result["bottom"] = p.bottom.Plan()
 	result["condition"] = p.cond.Plan()

sql3/planner/opnulltable.go

@@ -40,11 +40,7 @@ func (p *PlanOpNullTable) Children() []types.PlanOperator {
 func (p *PlanOpNullTable) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	return result
 }
 

sql3/planner/oporderby.go

@@ -86,11 +86,7 @@ func (n *PlanOpOrderBy) String() string {
 func (n *PlanOpOrderBy) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", n)
-	sc := make([]string, 0)
-	for _, e := range n.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
+	result["_schema"] = n.Schema().Plan()
 
 	result["child"] = n.ChildOp.Plan()
 	ps := make([]interface{}, 0)
@@ -150,7 +146,7 @@ func (i *orderByIter) Next(ctx context.Context) (types.Row, error) {
 }
 
 func (i *orderByIter) computeOrderByRows(ctx context.Context) error {
-	cache := newInMemoryRowCache()
+	cache := make([]types.Row, 0)
 
 	for {
 		row, err := i.childIter.Next(ctx)
@@ -162,15 +158,12 @@ func (i *orderByIter) computeOrderByRows(ctx context.Context) error {
 			return err
 		}
 
-		if err := cache.Add(row); err != nil {
-			return err
-		}
+		cache = append(cache, row)
 	}
 
-	rows := cache.AllRows()
 	sorter := &OrderBySorter{
 		SortFields: i.s.orderByFields,
-		Rows:       rows,
+		Rows:       cache,
 		LastError:  nil,
 		Ctx:        ctx,
 	}
@@ -178,7 +171,7 @@ func (i *orderByIter) computeOrderByRows(ctx context.Context) error {
 	if sorter.LastError != nil {
 		return sorter.LastError
 	}
-	i.sortedRows = rows
+	i.sortedRows = cache
 	return nil
 }
 

sql3/planner/oppqlaggregate.go

@@ -36,11 +36,7 @@ func NewPlanOpPQLAggregate(p *ExecutionPlanner, tableName string, aggregate type
 func (p *PlanOpPQLAggregate) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 	if p.filter != nil {
 		result["filter"] = p.filter.Plan()

sql3/planner/oppqldelete.go

@@ -33,11 +33,7 @@ func NewPlanOpPQLConstRowDelete(p *ExecutionPlanner, tableName string, child typ
 func (p *PlanOpPQLConstRowDelete) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["child"] = p.ChildOp.Plan()
 	result["tableName"] = p.tableName
 	return result

sql3/planner/oppqldistinctscan.go

@@ -0,0 +1,293 @@
+// Copyright 2023 Molecula Corp. All rights reserved.
+
+package planner
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	pilosa "github.com/molecula/featurebase/v3"
+	"github.com/molecula/featurebase/v3/dax"
+	"github.com/molecula/featurebase/v3/pql"
+	"github.com/molecula/featurebase/v3/sql3"
+	"github.com/molecula/featurebase/v3/sql3/parser"
+	"github.com/molecula/featurebase/v3/sql3/planner/types"
+	"github.com/pkg/errors"
+)
+
+// PlanOpPQLDistinctScan plan operator handles a PQL distinct scan
+// i.e. a scan with only one column that is used in a distinct query
+type PlanOpPQLDistinctScan struct {
+	planner   *ExecutionPlanner
+	tableName string
+	column    string
+	filter    types.PlanExpression
+	topExpr   types.PlanExpression
+	warnings  []string
+}
+
+func NewPlanOpPQLDistinctScan(p *ExecutionPlanner, tableName string, column string) (*PlanOpPQLDistinctScan, error) {
+	if strings.EqualFold("_id", column) {
+		return nil, sql3.NewErrInternalf("non _id column required")
+	}
+	return &PlanOpPQLDistinctScan{
+		planner:   p,
+		tableName: tableName,
+		column:    column,
+		warnings:  make([]string, 0),
+	}, nil
+}
+
+func (p *PlanOpPQLDistinctScan) Plan() map[string]interface{} {
+	result := make(map[string]interface{})
+	result["_op"] = fmt.Sprintf("%T", p)
+	result["_schema"] = p.Schema().Plan()
+	result["tableName"] = p.tableName
+
+	if p.topExpr != nil {
+		result["topExpr"] = p.topExpr.Plan()
+	}
+	if p.filter != nil {
+		result["filter"] = p.filter.Plan()
+	}
+	result["column"] = p.column
+	return result
+}
+
+func (p *PlanOpPQLDistinctScan) String() string {
+	return ""
+}
+
+func (p *PlanOpPQLDistinctScan) AddWarning(warning string) {
+	p.warnings = append(p.warnings, warning)
+}
+
+func (p *PlanOpPQLDistinctScan) Warnings() []string {
+	return p.warnings
+}
+
+func (p *PlanOpPQLDistinctScan) Name() string {
+	return p.tableName
+}
+
+func (p *PlanOpPQLDistinctScan) UpdateFilters(filterCondition types.PlanExpression) (types.PlanOperator, error) {
+	p.filter = filterCondition
+	return p, nil
+}
+
+func (p *PlanOpPQLDistinctScan) Schema() types.Schema {
+	result := make(types.Schema, 0)
+
+	tname := dax.TableName(p.tableName)
+	table, err := p.planner.schemaAPI.TableByName(context.Background(), tname)
+	if err != nil {
+		return result
+	}
+
+	for _, fld := range table.Fields {
+		if strings.EqualFold(string(fld.Name), p.column) {
+			result = append(result, &types.PlannerColumn{
+				ColumnName:   string(fld.Name),
+				RelationName: p.tableName,
+				Type:         fieldSQLDataType(pilosa.FieldToFieldInfo(fld)),
+			})
+			break
+		}
+	}
+	return result
+}
+
+func (p *PlanOpPQLDistinctScan) Children() []types.PlanOperator {
+	return []types.PlanOperator{}
+}
+
+func (p *PlanOpPQLDistinctScan) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
+	return &distinctScanRowIter{
+		planner:   p.planner,
+		tableName: p.tableName,
+		column:    p.column,
+		predicate: p.filter,
+		topExpr:   p.topExpr,
+	}, nil
+}
+
+func (p *PlanOpPQLDistinctScan) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
+	return nil, nil
+}
+
+type distinctScanRowIter struct {
+	planner   *ExecutionPlanner
+	tableName string
+	column    string
+	predicate types.PlanExpression
+	topExpr   types.PlanExpression
+
+	result         []interface{}
+	rowWidth       int
+	columnDataType parser.ExprDataType
+}
+
+var _ types.RowIterator = (*distinctScanRowIter)(nil)
+
+func (i *distinctScanRowIter) Next(ctx context.Context) (types.Row, error) {
+	if i.result == nil {
+		err := i.planner.checkAccess(ctx, i.tableName, accessTypeReadData)
+		if err != nil {
+			return nil, err
+		}
+
+		//go get the schema def and map names to indexes in the resultant row
+		tname := dax.TableName(i.tableName)
+		table, err := i.planner.schemaAPI.TableByName(context.Background(), tname)
+		if err != nil {
+			if errors.Is(err, pilosa.ErrIndexNotFound) {
+				return nil, sql3.NewErrInternalf("table not found '%s'", i.tableName)
+			}
+			return nil, err
+		}
+		i.rowWidth = 1
+
+		for _, fld := range table.Fields {
+			if strings.EqualFold(i.column, string(fld.Name)) {
+				i.columnDataType = fieldSQLDataType(pilosa.FieldToFieldInfo(fld))
+				break
+			}
+		}
+
+		var cond *pql.Call
+
+		cond, err = i.planner.generatePQLCallFromExpr(ctx, i.predicate)
+		if err != nil {
+			return nil, err
+		}
+		if cond == nil {
+			cond = &pql.Call{Name: "All"}
+		}
+
+		if i.topExpr != nil {
+			_, ok := i.topExpr.(*intLiteralPlanExpression)
+			if !ok {
+				return nil, sql3.NewErrInternalf("unexpected top expression type: %T", i.topExpr)
+			}
+			pqlValue, err := planExprToValue(i.topExpr)
+			if err != nil {
+				return nil, err
+			}
+			cond = &pql.Call{
+				Name:     "Limit",
+				Children: []*pql.Call{cond},
+				Args:     map[string]interface{}{"limit": pqlValue},
+				Type:     pql.PrecallGlobal,
+			}
+		}
+		call := &pql.Call{
+			Name:     "Distinct",
+			Args:     map[string]interface{}{"field": i.column},
+			Children: []*pql.Call{cond},
+		}
+
+		queryResponse, err := i.planner.executor.Execute(ctx, table, &pql.Query{Calls: []*pql.Call{call}}, nil, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		switch res := queryResponse.Results[0].(type) {
+		case *pilosa.Row:
+			result := make([]interface{}, 0)
+			if len(res.Keys) > 0 {
+				for _, n := range res.Keys {
+					result = append(result, n)
+				}
+			} else {
+				for _, n := range res.Columns() {
+					result = append(result, int64(n))
+				}
+			}
+			i.result = result
+
+		case pilosa.SignedRow:
+			result := make([]interface{}, 0)
+
+			negs := res.Neg.Columns()
+			pos := res.Pos.Columns()
+			for _, n := range negs {
+				result = append(result, -(int64(n)))
+			}
+			for _, n := range pos {
+				result = append(result, int64(n))
+			}
+			i.result = result
+
+		case pilosa.DistinctTimestamp:
+			result := make([]interface{}, 0)
+			for _, n := range res.Values {
+				if tm, err := time.ParseInLocation(time.RFC3339Nano, n, time.UTC); err == nil {
+					result = append(result, tm)
+				} else {
+					return nil, sql3.NewErrInternalf("unable to convert to time.Time: %v", n)
+				}
+			}
+			i.result = result
+
+		default:
+			return nil, sql3.NewErrInternalf("unexpected Distinct() result type: %T", res)
+		}
+	}
+
+	if len(i.result) > 0 {
+		row := make([]interface{}, i.rowWidth)
+
+		result := i.result[0]
+
+		switch t := i.columnDataType.(type) {
+
+		case *parser.DataTypeBool:
+			val, ok := result.(int64)
+			if !ok {
+				return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result)
+			}
+			row[0] = val == 1
+
+		case *parser.DataTypeDecimal:
+			val, ok := result.(int64)
+			if !ok {
+				return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result)
+			}
+			row[0] = pql.NewDecimal(val, t.Scale)
+
+		case *parser.DataTypeIDSet:
+			//empty sets are null
+			val, ok := result.([]uint64)
+			if !ok {
+				return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result)
+			}
+			if len(val) == 0 {
+				row[0] = nil
+			} else {
+				row[0] = val
+			}
+
+		case *parser.DataTypeStringSet:
+			//empty sets are null
+			val, ok := result.([]string)
+			if !ok {
+				return nil, sql3.NewErrInternalf("unexpected type for column value '%T'", result)
+			}
+			if len(val) == 0 {
+				row[0] = nil
+			} else {
+				row[0] = val
+			}
+
+		default:
+			row[0] = result
+		}
+
+		// Move to next result element.
+		i.result = i.result[1:]
+		return row, nil
+	}
+	return nil, types.ErrNoMoreRows
+}

sql3/planner/oppqlfiltereddelete.go

@@ -32,11 +32,7 @@ func NewPlanOpPQLFilteredDelete(p *ExecutionPlanner, tableName string, filter ty
 func (p *PlanOpPQLFilteredDelete) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["filter"] = p.filter.Plan()
 	result["tableName"] = p.tableName
 	return result

sql3/planner/oppqlgroupby.go

@@ -38,15 +38,10 @@ func NewPlanOpPQLGroupBy(p *ExecutionPlanner, tableName string, groupByExprs []t
 func (p *PlanOpPQLGroupBy) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 	if p.filter != nil {
 		result["filter"] = p.filter.Plan()
-
 	}
 	result["aggregate"] = p.aggregate.AggExpression().Plan()
 	ps := make([]interface{}, 0)

sql3/planner/oppqlmultiaggregate.go

@@ -27,12 +27,7 @@ func NewPlanOpPQLMultiAggregate(p *ExecutionPlanner, operators []*PlanOpPQLAggre
 func (p *PlanOpPQLMultiAggregate) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	ps := make([]interface{}, 0)
 	for _, e := range p.operators {
 		ps = append(ps, e.Plan())
@@ -59,7 +54,7 @@ func (p *PlanOpPQLMultiAggregate) Schema() types.Schema {
 		s := &types.PlannerColumn{
 			ColumnName:   aggOp.aggregate.String(),
 			RelationName: "",
-			Type:         aggOp.aggregate.AggExpression().Type(),
+			Type:         aggOp.aggregate.Type(),
 		}
 		result[idx] = s
 	}

sql3/planner/oppqlmultigroupby.go

@@ -33,12 +33,7 @@ func NewPlanOpPQLMultiGroupBy(p *ExecutionPlanner, operators []*PlanOpPQLGroupBy
 func (p *PlanOpPQLMultiGroupBy) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	ps := make([]interface{}, 0)
 	for _, e := range p.operators {
 		ps = append(ps, e.Plan())
@@ -135,9 +130,9 @@ type pqlMultiGroupByRowIter struct {
 	planner        *ExecutionPlanner
 	groupByColumns []types.PlanExpression
 	iterators      []types.RowIterator
-	groupCache     KeyedRowCache
+	groupCache     map[string]types.Row
 
-	groupKeys []uint64
+	groupKeys []string
 }
 
 var _ types.RowIterator = (*pqlMultiGroupByRowIter)(nil)
@@ -145,7 +140,7 @@ var _ types.RowIterator = (*pqlMultiGroupByRowIter)(nil)
 func (i *pqlMultiGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
 	if i.groupCache == nil {
 		//consume all the rows from the child iterators
-		i.groupCache = newinMemoryKeyedRowCache()
+		i.groupCache = make(map[string]types.Row)
 		if err := i.computeMultiGroupBy(ctx); err != nil {
 			return nil, err
 		}
@@ -154,9 +149,9 @@ func (i *pqlMultiGroupByRowIter) Next(ctx context.Context) (types.Row, error) {
 	if len(i.groupKeys) > 0 {
 		key := i.groupKeys[0]
 
-		row, err := i.groupCache.Get(key)
-		if err != nil {
-			return nil, err
+		row, ok := i.groupCache[key]
+		if !ok {
+			return nil, sql3.NewErrInternalf("unexpected absence of key")
 		}
 		// Move to next result element.
 		i.groupKeys = i.groupKeys[1:]
@@ -180,19 +175,16 @@ func (i *pqlMultiGroupByRowIter) computeMultiGroupBy(ctx context.Context) error
 
 		for {
 			//build a key for the group by columns for this row
-			key, _, err := groupingKeyHash(ctx, i.groupByColumns, irow)
-			if err != nil {
-				return err
-			}
-
-			// get the group from the cache
-			cachedRow, err := i.groupCache.Get(key)
+			key, _, err := groupingKey(ctx, i.groupByColumns, irow)
 			if err != nil {
 				return err
 			}
 
 			aggIndex := iteratorIdx + len(i.groupByColumns)
-			if cachedRow != nil {
+
+			// get the group from the cache
+			cachedRow, ok := i.groupCache[key]
+			if ok {
 				// if the group exists then update the row
 				// NB: the aggregate for this iterator is at the end of irow
 				cachedRow[aggIndex] = irow[len(irow)-1]
@@ -207,7 +199,7 @@ func (i *pqlMultiGroupByRowIter) computeMultiGroupBy(ctx context.Context) error
 				cachedRow[aggIndex] = irow[len(irow)-1]
 
 				// write the row to the cache
-				err = i.groupCache.Put(key, cachedRow)
+				i.groupCache[key] = cachedRow
 				if err != nil {
 					return err
 				}

sql3/planner/oppqltablescan.go

@@ -38,12 +38,7 @@ func NewPlanOpPQLTableScan(p *ExecutionPlanner, tableName string, columns []stri
 func (p *PlanOpPQLTableScan) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 
 	if p.topExpr != nil {
@@ -52,7 +47,6 @@ func (p *PlanOpPQLTableScan) Plan() map[string]interface{} {
 	if p.filter != nil {
 		result["filter"] = p.filter.Plan()
 	}
-
 	result["columns"] = p.columns
 	return result
 }

sql3/planner/opprojection.go

@@ -60,12 +60,8 @@ func (p *PlanOpProjection) WithChildren(children ...types.PlanOperator) (types.P
 
 func (p *PlanOpProjection) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
-	result["__op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["__schema"] = sc
+	result["_op"] = fmt.Sprintf("%T", p)
+	result["_schema"] = p.Schema().Plan()
 
 	result["child"] = p.ChildOp.Plan()
 
@@ -73,7 +69,7 @@ func (p *PlanOpProjection) Plan() map[string]interface{} {
 	for _, e := range p.Projections {
 		ps = append(ps, e.Plan())
 	}
-	result["_projections"] = ps
+	result["projections"] = ps
 
 	return result
 }

sql3/planner/opqltruncate.go

@@ -32,11 +32,7 @@ func NewPlanOpPQLTruncateTable(p *ExecutionPlanner, tableName string) *PlanOpPQL
 func (p *PlanOpPQLTruncateTable) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	result["tableName"] = p.tableName
 	return result
 }

sql3/planner/opquery.go

@@ -69,12 +69,7 @@ func (p *PlanOpQuery) WithChildren(children ...types.PlanOperator) (types.PlanOp
 func (p *PlanOpQuery) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	result["sql"] = p.sql
 	result["warnings"] = p.warnings
 	result["child"] = p.ChildOp.Plan()

sql3/planner/oprelalias.go

@@ -53,12 +53,7 @@ func (p *PlanOpRelAlias) WithChildren(children ...types.PlanOperator) (types.Pla
 func (p *PlanOpRelAlias) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	result["alias"] = p.alias
 	result["child"] = p.ChildOp.Plan()
 	return result

sql3/planner/opsubquery.go

@@ -46,12 +46,7 @@ func (p *PlanOpSubquery) WithChildren(children ...types.PlanOperator) (types.Pla
 func (p *PlanOpSubquery) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	result["child"] = p.ChildOp.Plan()
 	return result
 }

sql3/planner/opsystemtable.go

@@ -232,11 +232,7 @@ func NewPlanOpSystemTable(p *ExecutionPlanner, table *systemTable) *PlanOpSystem
 func (p *PlanOpSystemTable) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	ps := make([]string, 0)
-	for _, e := range p.Schema() {
-		ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = ps
+	result["_schema"] = p.Schema().Plan()
 	return result
 }
 

sql3/planner/optablevaluedfunction.go

@@ -57,12 +57,7 @@ func (p *PlanOpTableValuedFunction) WithChildren(children ...types.PlanOperator)
 func (p *PlanOpTableValuedFunction) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	return result
 }
 

sql3/planner/optop.go

@@ -53,12 +53,7 @@ func (p *PlanOpTop) WithChildren(children ...types.PlanOperator) (types.PlanOper
 func (p *PlanOpTop) Plan() map[string]interface{} {
 	result := make(map[string]interface{})
 	result["_op"] = fmt.Sprintf("%T", p)
-	sc := make([]string, 0)
-	for _, e := range p.Schema() {
-		sc = append(sc, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
-	}
-	result["_schema"] = sc
-
+	result["_schema"] = p.Schema().Plan()
 	result["expr"] = p.expr
 	result["child"] = p.ChildOp.Plan()
 	return result

sql3/planner/planoptimizer.go

@@ -26,6 +26,10 @@ var optimizerFunctions = []OptimizerFunc{
 	// fix expression references for having
 	removeUnusedExtractColumnReferences,
 
+	// if we have a distinct operator over a single projection,
+	// where the projection is on a table scan, use a PQL Distinct scan operator
+	tryToReplaceDistinctWithPQLDistinct,
+
 	// fix expression references for having
 	fixHavingReferences,
 
@@ -199,6 +203,8 @@ func getRelationAliases(n types.PlanOperator, scope *OptimizerScope) (RelationAl
 			switch t := node.ChildOp.(type) {
 			case *PlanOpPQLTableScan:
 				inspectErr = aliases.addAlias(node, t)
+			case *PlanOpPQLDistinctScan:
+				inspectErr = aliases.addAlias(node, t)
 			case *PlanOpSubquery:
 				inspectErr = aliases.addAlias(node, t)
 			default:
@@ -210,6 +216,10 @@ func getRelationAliases(n types.PlanOperator, scope *OptimizerScope) (RelationAl
 			inspectErr = aliases.addAlias(node, node)
 			return false
 
+		case *PlanOpPQLDistinctScan:
+			inspectErr = aliases.addAlias(node, node)
+			return false
+
 		}
 		return true
 	})
@@ -238,7 +248,7 @@ func filterPushdownAboveTablesChildSelector(c ParentContext) bool {
 	switch c.Parent.(type) {
 	case *PlanOpFilter:
 		switch c.Operator.(type) {
-		case *PlanOpRelAlias, *PlanOpPQLTableScan:
+		case *PlanOpRelAlias, *PlanOpPQLTableScan, *PlanOpPQLDistinctScan:
 			return false
 		}
 	}
@@ -339,6 +349,8 @@ func pushdownFiltersToFilterableRelations(ctx context.Context, a *ExecutionPlann
 	switch rel := tableNode.(type) {
 	case *PlanOpPQLTableScan:
 		table = rel
+	case *PlanOpPQLDistinctScan:
+		table = rel
 	default:
 		return tableNode, true, nil
 	}
@@ -391,6 +403,8 @@ func pushdownFiltersToAboveRelation(ctx context.Context, a *ExecutionPlanner, ta
 	switch rel := tableNode.(type) {
 	case *PlanOpPQLTableScan:
 		table = rel
+	case *PlanOpPQLDistinctScan:
+		table = rel
 	default:
 		return tableNode, true, nil
 	}
@@ -410,7 +424,7 @@ func pushdownFiltersToAboveRelation(ctx context.Context, a *ExecutionPlanner, ta
 	}
 
 	switch tableNode.(type) {
-	case *PlanOpRelAlias, *PlanOpPQLTableScan:
+	case *PlanOpRelAlias, *PlanOpPQLTableScan, *PlanOpPQLDistinctScan:
 		node := tableNode
 		if pushedDownFilterExpression != nil {
 			return NewPlanOpFilter(a, pushedDownFilterExpression, node), false, nil
@@ -442,7 +456,7 @@ func pushdownFilters(ctx context.Context, a *ExecutionPlanner, n types.PlanOpera
 				return n, samePred, nil
 
 			// PlanOpPQLTableScan supports being filtered, PlanOpRelAlias is included here as a "transparent" op
-			case *PlanOpRelAlias, *PlanOpPQLTableScan:
+			case *PlanOpRelAlias, *PlanOpPQLTableScan, *PlanOpPQLDistinctScan:
 				n, samePred, err := pushdownFiltersToFilterableRelations(ctx, a, node, scope, filters, tableAliases)
 				if err != nil {
 					return nil, true, err
@@ -468,7 +482,7 @@ func pushdownFilters(ctx context.Context, a *ExecutionPlanner, n types.PlanOpera
 				}
 				return n, false, nil
 
-			case *PlanOpRelAlias, *PlanOpPQLTableScan:
+			case *PlanOpRelAlias, *PlanOpPQLTableScan, *PlanOpPQLDistinctScan:
 				_, same, err := pushdownFiltersToAboveRelation(ctx, a, node, scope, filters)
 				if err != nil {
 					return nil, true, err
@@ -630,6 +644,91 @@ func tryToReplaceGroupByWithPQLAggregate(ctx context.Context, a *ExecutionPlanne
 	return n, true, nil
 }
 
+func tryToReplaceDistinctWithPQLDistinct(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
+	// bail if no distinct
+	hasDistinct := false
+	InspectPlan(n, func(node types.PlanOperator) bool {
+		switch node.(type) {
+		case *PlanOpDistinct:
+			hasDistinct = true
+			return false
+		}
+		return true
+	})
+	if !hasDistinct {
+		return n, true, nil
+	}
+
+	// bail if has a group by
+	hasGroupBy := false
+	InspectPlan(n, func(node types.PlanOperator) bool {
+		switch node.(type) {
+		case *PlanOpGroupBy:
+			hasGroupBy = true
+			return false
+		}
+		return true
+	})
+	if hasGroupBy {
+		return n, true, nil
+	}
+
+	//bail if there are any joins
+	joins, err := hasJoins(ctx, a, n, scope)
+	if err != nil {
+		return nil, false, err
+	}
+	if joins {
+		return n, true, nil
+	}
+
+	//go find the table scan operators
+	tables := getTableScanOperators(ctx, a, n, scope)
+
+	//only do this if we have one TableScanOperator
+	if len(tables) == 1 {
+		replacedWithDistinct := false
+		// replace the scan with the distinct scan
+		return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) {
+			switch thisNode := node.(type) {
+			case *PlanOpDistinct:
+				if replacedWithDistinct {
+					return thisNode.ChildOp, false, nil
+				}
+				return thisNode, true, nil
+
+			case *PlanOpPQLTableScan:
+				// bail if there is more than one output column
+				if len(thisNode.columns) != 1 {
+					return thisNode, true, nil
+				}
+
+				// make sure it's not the _id column
+				if strings.EqualFold(thisNode.columns[0], "_id") {
+					return thisNode, true, nil
+				}
+
+				// make sure it's not a set type
+				s := thisNode.Schema()
+				switch s[0].Type.(type) {
+				case *parser.DataTypeIDSet, *parser.DataTypeStringSet:
+					return thisNode, true, nil
+				}
+
+				newOp, err := NewPlanOpPQLDistinctScan(a, thisNode.tableName, thisNode.columns[0])
+				if err != nil {
+					return nil, false, err
+				}
+				replacedWithDistinct = true
+				return newOp, false, nil
+			default:
+				return thisNode, true, nil
+			}
+		})
+	}
+	return n, true, nil
+}
+
 func tryToReplaceConstRowDeleteWithFilteredDelete(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (types.PlanOperator, bool, error) {
 	return TransformPlanOp(n, func(node types.PlanOperator) (types.PlanOperator, bool, error) {
 		switch node := node.(type) {
@@ -881,7 +980,7 @@ func fixProjectionReferences(ctx context.Context, a *ExecutionPlanner, n types.P
 				return thisNode, false, nil
 
 			// everything else that can be a child of projection
-			case *PlanOpRelAlias, *PlanOpFilter, *PlanOpPQLTableScan, *PlanOpNestedLoops:
+			case *PlanOpRelAlias, *PlanOpFilter, *PlanOpPQLTableScan, *PlanOpPQLDistinctScan, *PlanOpNestedLoops:
 				exprs, same, err := fixFieldRefIndexesOnExpressions(ctx, scope, a, childOp.Schema(), thisNode.Projections...)
 				if err != nil {
 					return thisNode, true, err
@@ -999,28 +1098,6 @@ func fixHavingReferences(ctx context.Context, a *ExecutionPlanner, n types.PlanO
 	})
 }
 
-// hasTop inspects a plan op tree and returns true (or error) if there are Top
-// operators.
-func hasTop(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (bool, error) {
-	result := false
-	InspectPlan(n, func(node types.PlanOperator) bool {
-		switch node.(type) {
-		case *PlanOpTop:
-			result = true
-			return false
-		}
-		return true
-	})
-	return result, nil
-}
-
-// hasTopN inspects a plan op tree and returns true (or error) if there are TopN
-// operators.
-func hasTopN(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (bool, error) {
-	//TODO(pok) implement this
-	return false, nil
-}
-
 // inspects a plan op tree and returns false (or error) if there are read join operators
 func hasJoins(ctx context.Context, a *ExecutionPlanner, n types.PlanOperator, scope *OptimizerScope) (bool, error) {
 	// assume false

sql3/planner/types/operator.go

@@ -67,12 +67,26 @@ type FilteredRelation interface {
 // Schema is the definition a set of columns from each operator
 type Schema []*PlannerColumn
 
+func (r Schema) Plan() []map[string]interface{} {
+	result := make([]map[string]interface{}, len(r))
+	for i, s := range r {
+		m := make(map[string]interface{})
+		m["name"] = s.ColumnName
+		m["alias"] = s.AliasName
+		m["relation"] = s.RelationName
+		m["type"] = s.Type.TypeDescription()
+		result[i] = m
+	}
+	return result
+}
+
 // Row is a tuple of values
 type Row []interface{}
 
 // Append appends all the values in r2 to this row and returns the result
 func (r Row) Append(r2 Row) Row {
 	row := make(Row, len(r)+len(r2))
+	// TODO(pok) use a copy here
 	for i := range r {
 		row[i] = r[i]
 	}

sql3/planner/types/planexpression.go

@@ -59,6 +59,7 @@ type Aggregable interface {
 	AggType() AggregateFunctionType
 	AggExpression() PlanExpression
 	AggAdditionalExpr() []PlanExpression
+	Type() parser.ExprDataType
 }
 
 // interface to something that can be identified by a name

sql3/test/defs/defs.go

@@ -17,6 +17,7 @@ var TableTests []TableTest = []TableTest{
 	selectKeyedTests,
 	selectHavingTests,
 	orderByTests,
+	distinctTests,
 
 	topTests,
 

sql3/test/defs/defs_distinct.go

@@ -0,0 +1,157 @@
+package defs
+
+import "github.com/molecula/featurebase/v3/pql"
+
+// distinct tests
+var distinctTests = TableTest{
+	Table: tbl(
+		"distinct_test",
+		srcHdrs(
+			srcHdr("_id", fldTypeID),
+			srcHdr("i1", fldTypeInt),
+			srcHdr("b1", fldTypeBool),
+			srcHdr("id1", fldTypeID),
+			srcHdr("ids1", fldTypeIDSet),
+			srcHdr("d1", fldTypeDecimal2),
+			srcHdr("s1", fldTypeString),
+			srcHdr("ss1", fldTypeStringSet),
+			srcHdr("ts1", fldTypeTimestamp),
+		),
+		srcRows(
+			srcRow(int64(1), int64(10), bool(false), int64(1), []int64{10, 20, 30}, float64(10.00), string("10"), []string{"10", "20", "30"}, knownTimestamp()),
+			srcRow(int64(2), int64(20), bool(true), int64(2), []int64{11, 21, 31}, float64(20.00), string("20"), []string{"11", "21", "31"}, knownTimestamp()),
+			srcRow(int64(3), int64(30), bool(false), int64(3), []int64{12, 22, 32}, float64(30.00), string("30"), []string{"12", "22", "32"}, knownTimestamp()),
+			srcRow(int64(4), int64(10), bool(false), int64(1), []int64{10, 20, 30}, float64(10.00), string("10"), []string{"10", "20", "30"}, knownTimestamp()),
+			srcRow(int64(5), int64(20), bool(true), int64(2), []int64{11, 21, 31}, float64(20.00), string("20"), []string{"11", "21", "31"}, knownTimestamp()),
+			srcRow(int64(6), int64(30), bool(false), int64(3), []int64{12, 22, 32}, float64(30.00), string("30"), []string{"12", "22", "32"}, knownTimestamp()),
+		),
+	),
+	SQLTests: []SQLTest{
+		{
+			SQLs: sqls(
+				"select distinct i1, b1, id1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("i1", fldTypeInt),
+				hdr("b1", fldTypeBool),
+				hdr("id1", fldTypeID),
+			),
+			ExpRows: rows(
+				row(int64(10), bool(false), int64(1)),
+				row(int64(20), bool(true), int64(2)),
+				row(int64(30), bool(false), int64(3)),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct i1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("i1", fldTypeInt),
+			),
+			ExpRows: rows(
+				row(int64(10)),
+				row(int64(20)),
+				row(int64(30)),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct b1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("b1", fldTypeBool),
+			),
+			ExpRows: rows(
+				row(bool(false)),
+				row(bool(true)),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct id1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("id1", fldTypeID),
+			),
+			ExpRows: rows(
+				row(int64(1)),
+				row(int64(2)),
+				row(int64(3)),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct ids1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("ids1", fldTypeIDSet),
+			),
+			ExpRows: rows(
+				row([]int64{10, 20, 30}),
+				row([]int64{11, 21, 31}),
+				row([]int64{12, 22, 32}),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct d1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("d1", fldTypeDecimal2),
+			),
+			ExpRows: rows(
+				row(pql.NewDecimal(1000, 2)),
+				row(pql.NewDecimal(2000, 2)),
+				row(pql.NewDecimal(3000, 2)),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct s1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("s1", fldTypeString),
+			),
+			ExpRows: rows(
+				row(string("10")),
+				row(string("20")),
+				row(string("30")),
+			),
+			Compare: CompareExactUnordered,
+		},
+		{
+			SQLs: sqls(
+				"select distinct ss1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("ss1", fldTypeStringSet),
+			),
+			ExpRows: rows(
+				row([]string{"10", "20", "30"}),
+				row([]string{"11", "21", "31"}),
+				row([]string{"12", "22", "32"}),
+			),
+			Compare:        CompareExactUnordered,
+			SortStringKeys: true,
+		},
+		{
+			SQLs: sqls(
+				"select distinct ts1 from distinct_test",
+			),
+			ExpHdrs: hdrs(
+				hdr("ts1", fldTypeTimestamp),
+			),
+			ExpRows: rows(
+				row(knownTimestamp()),
+			),
+			Compare: CompareExactUnordered,
+		},
+	},
+}