Nested query using = should return error message if there's more than one row returned

[jkosh44]

Bug Report

Summary

When using a nested query in the WHERE clause you can use an operator such as = or < as long as the nested query returns a single row. However if the nested query returns more than one row than we should return an error to the user. Currently NoisePage crashes instead of returning an error message.

Environment

OS: Ubuntu (LTS) 20.04

Compiler: GCC 7.0+

CMake Profile: Debug

Jenkins/CI: N/A

Steps to Reproduce

noisepage=# CREATE TABLE foo1 (a int);
CREATE TABLE
noisepage=# CREATE TABLE foo2 (a int);
CREATE TABLE
noisepage=# INSERT INTO foo1 VALUES (1);
INSERT 0 1
noisepage=# INSERT INTO foo1 VALUES (2);
INSERT 0 1
noisepage=# INSERT INTO foo2 VALUES (1);
INSERT 0 1
noisepage=# INSERT INTO foo2 VALUES (2);
INSERT 0 1
noisepage=# SELECT * FROM foo1 WHERE a = (SELECT a FROM foo2);
server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.
The connection to the server was lost. Attempting reset: Failed.
!> 

The DB crashes with a Segmentation Fault

Expected Behavior

postgres=# CREATE TABLE foo1 (a int);
CREATE TABLE
postgres=# CREATE TABLE foo2 (a int);
CREATE TABLE
postgres=# INSERT INTO foo1 VALUES (1);
INSERT 0 1
postgres=# INSERT INTO foo1 VALUES (2);
INSERT 0 1
postgres=# INSERT INTO foo2 VALUES (1);
INSERT 0 1
postgres=# INSERT INTO foo2 VALUES (2);
INSERT 0 1
postgres=# SELECT * FROM foo1 WHERE a = (SELECT a FROM foo2);
ERROR:  more than one row returned by a subquery used as an expression

Note: the DB doesn't crash from this, just prints an error to the terminal

Notes

There are other similar queries that cause errors (They all start with the same creates and inserts). In Postgres they all have the same behavior of printing the same error message to the terminal.

• SELECT * FROM foo1 WHERE a > (SELECT a FROM foo2); Also causes a segfault
• SELECT * FROM foo1 WHERE a = (SELECT MIN(a) FROM foo2 GROUP BY a); Errors out and prints the following to the console:

[2020-12-03 20:30:59.532] [execution_logger] [error] ERROR: Line: 62, Col: 0 => function 'aggHTIterInit' expects argument of type '*AggregationHashTable' in position '1', received type '*TableVectorIterator'

struct OutputStruct {
    out0: Integer
}
struct AggPayload {
    gb_term_attr0 : Integer
    agg_term_attr0: IntegerMinAggregate
}
struct AggValues {
    gb_term_attr0 : Integer
    agg_term_attr0: Integer
}
struct QueryState {
    execCtx     : *ExecutionContext
    aggHashTable: AggregationHashTable
}
struct P2_State {
    aggHashTable1: AggregationHashTable
}
struct P1_State {
    output_buffer: *OutputBuffer
}
fun Query4_Pipeline1_KeyCheckPartial(lhs: *AggPayload, rhs: *AggPayload) -> bool {
    if (SqlBoolToBool(lhs.gb_term_attr0 != rhs.gb_term_attr0)) {
        return false
    }
    return true
}

fun Query4_Pipeline1_MergePartitions(queryState: *QueryState, aggHashTable: *AggregationHashTable, ahtOvfIter: *AHTOverflowPartitionIterator) -> nil {
    for (; @aggPartIterHasNext(ahtOvfIter); @aggPartIterNext(ahtOvfIter)) {
        var hashVal = @aggPartIterGetHash(ahtOvfIter)
        var partialRow = @ptrCast(*AggPayload, @aggPartIterGetRow(ahtOvfIter))
        var aggPayload = @ptrCast(*AggPayload, @aggHTLookup(aggHashTable, hashVal, Query4_Pipeline1_KeyCheckPartial, partialRow))
        if (aggPayload == nil) {
            @aggHTLink(aggHashTable, @aggPartIterGetRowEntry(ahtOvfIter))
        } else {
            @aggMerge(&aggPayload.agg_term_attr0, &partialRow.agg_term_attr0)
        }
    }
    return
}

fun Query4_Pipeline1_KeyCheck(aggPayload: *AggPayload, aggValues: *AggValues) -> bool {
    if (SqlBoolToBool(aggPayload.gb_term_attr0 != aggValues.gb_term_attr0)) {
        return false
    }
    return true
}

fun Query4_Init(queryState: *QueryState) -> nil {
    @aggHTInit(&queryState.aggHashTable, queryState.execCtx, @sizeOf(AggPayload))
    return
}

fun Query4_Pipeline2_InitPipelineState(queryState: *QueryState, pipelineState: *P2_State) -> nil {
    @aggHTInit(&pipelineState.aggHashTable1, queryState.execCtx, @sizeOf(AggPayload))
    return
}

fun Query4_Pipeline2_TearDownPipelineState(queryState: *QueryState, pipelineState: *P2_State) -> nil {
    @aggHTFree(&pipelineState.aggHashTable1)
    return
}

fun Query4_Pipeline2_ParallelWork(queryState: *QueryState, pipelineState: *P2_State, tvi: *TableVectorIterator) -> nil {
    var slot: TupleSlot
    for (@tableIterAdvance(tvi)) {
        var vpi = @tableIterGetVPI(tvi)
        for (; @vpiHasNext(vpi); @vpiAdvance(vpi)) {
            slot = @vpiGetSlot(vpi)
            var aggValues: AggValues
            aggValues.gb_term_attr0 = @vpiGetIntNull(vpi, 0)
            aggValues.agg_term_attr0 = @vpiGetIntNull(vpi, 0)
            var hashVal = @hash(aggValues.gb_term_attr0)
            var aggPayload = @ptrCast(*AggPayload, @aggHTLookup(&pipelineState.aggHashTable1, hashVal, Query4_Pipeline1_KeyCheck, &aggValues))
            if (aggPayload == nil) {
                aggPayload = @ptrCast(*AggPayload, @aggHTInsert(&pipelineState.aggHashTable1, hashVal, true))
                aggPayload.gb_term_attr0 = aggValues.gb_term_attr0
                @aggInit(&aggPayload.agg_term_attr0)
            }
            @aggAdvance(&aggPayload.agg_term_attr0, &aggValues.agg_term_attr0)
        }
        var vpi_num_tuples = @tableIterGetVPINumTuples(tvi)
    }
    return
}

fun Query4_Pipeline2_Init(queryState: *QueryState) -> nil {
    var threadStateContainer = @execCtxGetTLS(queryState.execCtx)
    @tlsReset(threadStateContainer, @sizeOf(P2_State), Query4_Pipeline2_InitPipelineState, Query4_Pipeline2_TearDownPipelineState, queryState)
    return
}

fun Query4_Pipeline2_Run(queryState: *QueryState) -> nil {
    var pipelineState = @ptrCast(*P2_State, @tlsGetCurrentThreadState(@execCtxGetTLS(queryState.execCtx)))
    var col_oids: [1]uint32
    col_oids[0] = 1
    @iterateTableParallel(1061, col_oids, queryState, queryState.execCtx, Query4_Pipeline2_ParallelWork)
    @aggHTMoveParts(&queryState.aggHashTable, @execCtxGetTLS(queryState.execCtx), @offsetOf(P2_State, aggHashTable1), Query4_Pipeline1_MergePartitions)
    return
}

fun Query4_Pipeline2_TearDown(queryState: *QueryState) -> nil {
    @tlsClear(@execCtxGetTLS(queryState.execCtx))
    @ensureTrackersStopped(queryState.execCtx)
    return
}

fun Query4_Pipeline1_InitPipelineState(queryState: *QueryState, pipelineState: *P1_State) -> nil {
    pipelineState.output_buffer = @resultBufferNew(queryState.execCtx)
    return
}

fun Query4_Pipeline1_TearDownPipelineState(queryState: *QueryState, pipelineState: *P1_State) -> nil {
    @resultBufferFree(pipelineState.output_buffer)
    return
}

fun Query4_Pipeline1_ParallelWork(queryState: *QueryState, pipelineState: *P1_State, tvi1: *TableVectorIterator) -> nil {
    var slot1: TupleSlot
    for (@tableIterAdvance(tvi1)) {
        var vpi1 = @tableIterGetVPI(tvi1)
        for (; @vpiHasNext(vpi1); @vpiAdvance(vpi1)) {
            slot1 = @vpiGetSlot(vpi1)
            var iterBase: AHTIterator
            var iter = &iterBase
            for (@aggHTIterInit(iter, tvi1); @aggHTIterHasNext(iter); @aggHTIterNext(iter)) {
                var aggRow = @ptrCast(*AggPayload, @aggHTIterGetRow(iter))
                if (SqlBoolToBool(@vpiGetIntNull(vpi1, 0) == @aggResult(&aggRow.agg_term_attr0))) {
                    var outRow = @ptrCast(*OutputStruct, @resultBufferAllocRow(pipelineState.output_buffer))
                    outRow.out0 = @vpiGetIntNull(vpi1, 0)
                }
            }
            @aggHTIterClose(iter)
        }
        var vpi_num_tuples = @tableIterGetVPINumTuples(tvi1)
    }
    @resultBufferFinalize(pipelineState.output_buffer)
    return
}

fun Query4_Pipeline1_Init(queryState: *QueryState) -> nil {
    var threadStateContainer = @execCtxGetTLS(queryState.execCtx)
    @tlsReset(threadStateContainer, @sizeOf(P1_State), Query4_Pipeline1_InitPipelineState, Query4_Pipeline1_TearDownPipelineState, queryState)
    return
}

fun Query4_Pipeline1_Run(queryState: *QueryState) -> nil {
    var pipelineState = @ptrCast(*P1_State, @tlsGetCurrentThreadState(@execCtxGetTLS(queryState.execCtx)))
    var col_oids1: [1]uint32
    col_oids1[0] = 1
    @iterateTableParallel(1060, col_oids1, queryState, queryState.execCtx, Query4_Pipeline1_ParallelWork)
    @resultBufferFinalize(pipelineState.output_buffer)
    return
}

fun Query4_Pipeline1_TearDown(queryState: *QueryState) -> nil {
    @tlsClear(@execCtxGetTLS(queryState.execCtx))
    @ensureTrackersStopped(queryState.execCtx)
    return
}

fun Query4_TearDown(queryState: *QueryState) -> nil {
    @aggHTFree(&queryState.aggHashTable)
    return
}


noisepage: ../src/execution/compiler/executable_query.cpp:139: void noisepage::execution::compiler::ExecutableQuery::Setup(std::vector<std::unique_ptr<noisepage::execution::compiler::ExecutableQuery::Fragment> >&&, std::size_t, std::unique_ptr<noisepage::selfdriving::PipelineOperatingUnits>): Assertion `(std::all_of(fragments.begin(), fragments.end(), [](const auto &fragment) { return fragment->IsCompiled(); })) && ("All query fragments are not compiled!")' failed.
Signal: SIGABRT (Aborted)