diff --git a/bufferpool/bufferpool.go b/bufferpool/bufferpool.go new file mode 100644 index 000000000..5cd3f0f72 --- /dev/null +++ b/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() +} diff --git a/bufferpool/circularlist.go b/bufferpool/circularlist.go new file mode 100644 index 000000000..696e7f438 --- /dev/null +++ b/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-- +} diff --git a/bufferpool/clockreplacer.go b/bufferpool/clockreplacer.go new file mode 100644 index 000000000..4210c5f73 --- /dev/null +++ b/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 = ¤tNode.next + } else { + frameID := currentNode.key.(FrameID) + victimFrameID = frameID + c.clockHand = ¤tNode.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 +} diff --git a/bufferpool/diskmanager.go b/bufferpool/diskmanager.go new file mode 100644 index 000000000..6eebcdc10 --- /dev/null +++ b/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() +} diff --git a/bufferpool/inmemdiskmanager.go b/bufferpool/inmemdiskmanager.go new file mode 100644 index 000000000..83b26d8a6 --- /dev/null +++ b/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()) + } +} diff --git a/bufferpool/page.go b/bufferpool/page.go new file mode 100644 index 000000000..411699402 --- /dev/null +++ b/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) | +// |====================================================| +// | | +// |----------------------------------------------------| +// | 20 | slotcount | slot entry is 2 int16 | +// | | * slotwidth | values (payloadOffset, | +// | | * #slots | payloadLength) | +// |----------------------------------------------------| +// | | +// |----------------------------------------------------| +// | | +// | 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) + } + +} diff --git a/dax/queryer/orchestrator.go b/dax/queryer/orchestrator.go index f3151972b..13562e736 100644 --- a/dax/queryer/orchestrator.go +++ b/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 { diff --git a/dax/table.go b/dax/table.go index 3c23f511d..eb39df22a 100644 --- a/dax/table.go +++ b/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. diff --git a/extendiblehash/extendiblehash.go b/extendiblehash/extendiblehash.go new file mode 100644 index 000000000..7cd449961 --- /dev/null +++ b/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 +} diff --git a/extendiblehash/extendiblehash_test.go b/extendiblehash/extendiblehash_test.go new file mode 100644 index 000000000..dc299846f --- /dev/null +++ b/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 +} diff --git a/extendiblehash/key.go b/extendiblehash/key.go new file mode 100644 index 000000000..ec7f55da5 --- /dev/null +++ b/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 +} diff --git a/go.mod b/go.mod index 1dbef6739..8fd3966df 100644 --- a/go.mod +++ b/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 diff --git a/sql3/planner/compileselect.go b/sql3/planner/compileselect.go index f2fca77c7..abdca8222 100644 --- a/sql3/planner/compileselect.go +++ b/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 diff --git a/sql3/planner/expression.go b/sql3/planner/expression.go index d429a3842..fc69a7798 100644 --- a/sql3/planner/expression.go +++ b/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()) diff --git a/sql3/planner/expressionagg.go b/sql3/planner/expressionagg.go index 9eeabe22a..6bd90ed86 100644 --- a/sql3/planner/expressionagg.go +++ b/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() diff --git a/sql3/planner/memoryobj.go b/sql3/planner/memoryobj.go deleted file mode 100644 index 784127149..000000000 --- a/sql3/planner/memoryobj.go +++ /dev/null @@ -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{}), - } -} diff --git a/sql3/planner/opbulkinsert.go b/sql3/planner/opbulkinsert.go index 742754a8e..6dcf19e16 100644 --- a/sql3/planner/opbulkinsert.go +++ b/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{}) diff --git a/sql3/planner/opcreatetable.go b/sql3/planner/opcreatetable.go index 3414e44e3..122dcc7df 100644 --- a/sql3/planner/opcreatetable.go +++ b/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 diff --git a/sql3/planner/opdistinct.go b/sql3/planner/opdistinct.go index 288ff6346..374fb6207 100644 --- a/sql3/planner/opdistinct.go +++ b/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() +} diff --git a/sql3/planner/opdroptable.go b/sql3/planner/opdroptable.go index a25891ea5..f611478ea 100644 --- a/sql3/planner/opdroptable.go +++ b/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 } diff --git a/sql3/planner/opfeaturebasecolumns.go b/sql3/planner/opfeaturebasecolumns.go index 9e8aa23c3..d0d73b5ee 100644 --- a/sql3/planner/opfeaturebasecolumns.go +++ b/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 } diff --git a/sql3/planner/opfeaturebasetables.go b/sql3/planner/opfeaturebasetables.go index 1268a815a..302039bb7 100644 --- a/sql3/planner/opfeaturebasetables.go +++ b/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 } diff --git a/sql3/planner/opfilter.go b/sql3/planner/opfilter.go index b63228040..37a249f7b 100644 --- a/sql3/planner/opfilter.go +++ b/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 diff --git a/sql3/planner/opgroupby.go b/sql3/planner/opgroupby.go index aac4c811b..177b2b19a 100644 --- a/sql3/planner/opgroupby.go +++ b/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 } diff --git a/sql3/planner/ophaving.go b/sql3/planner/ophaving.go index 44fa57247..83d092406 100644 --- a/sql3/planner/ophaving.go +++ b/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 diff --git a/sql3/planner/opinsert.go b/sql3/planner/opinsert.go index 5819bb2fe..93cdda4a4 100644 --- a/sql3/planner/opinsert.go +++ b/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 } diff --git a/sql3/planner/opnestedloops.go b/sql3/planner/opnestedloops.go index 7fe91dcca..71df5894b 100644 --- a/sql3/planner/opnestedloops.go +++ b/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() diff --git a/sql3/planner/opnulltable.go b/sql3/planner/opnulltable.go index f6a861106..304167135 100644 --- a/sql3/planner/opnulltable.go +++ b/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 } diff --git a/sql3/planner/oporderby.go b/sql3/planner/oporderby.go index c9a6b6c9b..b11c431a1 100644 --- a/sql3/planner/oporderby.go +++ b/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 } diff --git a/sql3/planner/oppqlaggregate.go b/sql3/planner/oppqlaggregate.go index 82938b801..8756add7b 100644 --- a/sql3/planner/oppqlaggregate.go +++ b/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() diff --git a/sql3/planner/oppqldelete.go b/sql3/planner/oppqldelete.go index 1c84706ff..8a610050c 100644 --- a/sql3/planner/oppqldelete.go +++ b/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 diff --git a/sql3/planner/oppqldistinctscan.go b/sql3/planner/oppqldistinctscan.go new file mode 100644 index 000000000..51ddfc7a3 --- /dev/null +++ b/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 +} diff --git a/sql3/planner/oppqlfiltereddelete.go b/sql3/planner/oppqlfiltereddelete.go index 0e48ff34d..75aeac0b1 100644 --- a/sql3/planner/oppqlfiltereddelete.go +++ b/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 diff --git a/sql3/planner/oppqlgroupby.go b/sql3/planner/oppqlgroupby.go index 532421a28..2afbd3dbb 100644 --- a/sql3/planner/oppqlgroupby.go +++ b/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) diff --git a/sql3/planner/oppqlmultiaggregate.go b/sql3/planner/oppqlmultiaggregate.go index 53712fdf0..634cfa185 100644 --- a/sql3/planner/oppqlmultiaggregate.go +++ b/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 } diff --git a/sql3/planner/oppqlmultigroupby.go b/sql3/planner/oppqlmultigroupby.go index 0719dbc7c..3ad787b2e 100644 --- a/sql3/planner/oppqlmultigroupby.go +++ b/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 } diff --git a/sql3/planner/oppqltablescan.go b/sql3/planner/oppqltablescan.go index 539044eb5..542b513a2 100644 --- a/sql3/planner/oppqltablescan.go +++ b/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 } diff --git a/sql3/planner/opprojection.go b/sql3/planner/opprojection.go index 32f1bc231..8aee7a879 100644 --- a/sql3/planner/opprojection.go +++ b/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 } diff --git a/sql3/planner/opqltruncate.go b/sql3/planner/opqltruncate.go index c479bafc4..a3ebd7ec4 100644 --- a/sql3/planner/opqltruncate.go +++ b/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 } diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index e68ed8e5f..6b34a2327 100644 --- a/sql3/planner/opquery.go +++ b/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() diff --git a/sql3/planner/oprelalias.go b/sql3/planner/oprelalias.go index 98fd7aa41..2e4d5610c 100644 --- a/sql3/planner/oprelalias.go +++ b/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 diff --git a/sql3/planner/opsubquery.go b/sql3/planner/opsubquery.go index 6425012d4..105815c9c 100644 --- a/sql3/planner/opsubquery.go +++ b/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 } diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go index f7a983c27..a0fb33431 100644 --- a/sql3/planner/opsystemtable.go +++ b/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 } diff --git a/sql3/planner/optablevaluedfunction.go b/sql3/planner/optablevaluedfunction.go index 9f7ff23d0..1619fcdac 100644 --- a/sql3/planner/optablevaluedfunction.go +++ b/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 } diff --git a/sql3/planner/optop.go b/sql3/planner/optop.go index 2d6814c87..0e7d44269 100644 --- a/sql3/planner/optop.go +++ b/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 diff --git a/sql3/planner/planoptimizer.go b/sql3/planner/planoptimizer.go index 2e46ea846..cac459f8c 100644 --- a/sql3/planner/planoptimizer.go +++ b/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 diff --git a/sql3/planner/types/operator.go b/sql3/planner/types/operator.go index dac15002c..ef36f1650 100644 --- a/sql3/planner/types/operator.go +++ b/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] } diff --git a/sql3/planner/types/planexpression.go b/sql3/planner/types/planexpression.go index ad99a15a4..49e6a6069 100644 --- a/sql3/planner/types/planexpression.go +++ b/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 diff --git a/sql3/test/defs/defs.go b/sql3/test/defs/defs.go index 162d94f3b..3018951b4 100644 --- a/sql3/test/defs/defs.go +++ b/sql3/test/defs/defs.go @@ -17,6 +17,7 @@ var TableTests []TableTest = []TableTest{ selectKeyedTests, selectHavingTests, orderByTests, + distinctTests, topTests, diff --git a/sql3/test/defs/defs_distinct.go b/sql3/test/defs/defs_distinct.go new file mode 100644 index 000000000..8e425d597 --- /dev/null +++ b/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, + }, + }, +}