关于数据库:浪潮云溪分布式数据库协议代码解析2

– 数据申请阶段 –

Part 1 – 简略查问

客户端发送 Query (‘Q’) 音讯给服务端，蕴含了一条字符串类型的 SQL 语句。

func(cn *conn) query(query string, args []driver.Value)(_ *rows, err error){
    ...
    // Check to see if we can use the "simpleQuery"interface, which is
    // *much* faster than going through prepare/exec
    iflen(args)==0{return cn.simpleQuery(query)
    }
    ...
}

服务端收到 Query 音讯，解析 SQL 语句，生成形象语法树 (AST)，并传给执行器执行，取得后果。

func(c *conn) serveImpl(
    ctx context.Context,
    draining func()bool,
    sqlServer *sql.Server,
    reserved mon.BoundAccount,
    stopper *stop.Stopper,
)error{
    ...
Loop:
    for{typ, n, err = c.readBuf.ReadTypedMsg(&c.rd)
        if err !=nil{break Loop}
        ...
        switch typ {
        case pgwirebase.ClientMsgSimpleQuery:
            ...
        case pgwirebase.ClientMsgExecute:
            ...
        case pgwirebase.ClientMsgParse:
            ...
        case pgwirebase.ClientMsgDescribe:
            ...
        case pgwirebase.ClientMsgBind:
            ...
        case pgwirebase.ClientMsgSync:
            ...
        }
    }
    ...
}

服务端依据 SQL 后果，首先发送 RowDescription(B:‘T’) 音讯，蕴含列的数量，列名，列的类型等参数。

func(c *conn) writeRowDescription(
    ctx context.Context,
    columns []sqlbase.ResultColumn,
    formatCodes []pgwirebase.FormatCode,
    w io.Writer,
)error{c.msgBuilder.initMsg(pgwirebase.ServerMsgRowDescription)
    c.msgBuilder.putInt16(int16(len(columns)))
    for i, column :=range columns {
        ...
        c.msgBuilder.writeTerminatedString(column.Name)
        ...
        c.msgBuilder.putInt32(0)//Table OID (optional).
        c.msgBuilder.putInt16(0)//Column attribute ID (optional).
        c.msgBuilder.putInt32(int32(typ.oid))
        c.msgBuilder.putInt16(int16(typ.size))
        ...
    }
    ...
}

RowDescription 音讯前面将跟着多个 DataRow(B:‘D’) 音讯，每个 DataRow 音讯蕴含一行的数据。

func(c *conn) bufferRow(
    ctx context.Context,
    row tree.Datums,
    formatCodes []pgwirebase.FormatCode,
    convsessiondata.DataConversionConfig,
    types []*types.T,){c.msgBuilder.initMsg(pgwirebase.ServerMsgDataRow)
    c.msgBuilder.putInt16(int16(len(row)))
    for i, col :=range row {
        ...
        switch fmtCode {
        case pgwirebase.FormatText:
            c.msgBuilder.writeTextDatum(ctx, col, conv, types[i])
        case pgwirebase.FormatBinary:
            c.msgBuilder.writeBinaryDatum(ctx, col, conv.Location, types[i])
        ...
    }
    if err := c.msgBuilder.finishMsg(&c.writerState.buf); err !=nil{panic(fmt.Sprintf("unexpected err from buffer: %s", err))
    }
}

发送 CommandComplete(B:‘C’) 音讯示意这个 SQL 申请执行完结了。
服务端发送 ReadyForQuery(‘Z’)，告诉客户端能够发送下一条 SQL 申请了。

func(r *commandResult) Close(ctx context.Context, t sql.TransactionStatusIndicator){
    ...
    switch r.typ {
    case commandComplete:
        tag := cookTag(r.cmdCompleteTag, r.conn.writerState.tagBuf[:0], r.stmtType, r.rowsAffected,
        )
        r.conn.bufferCommandComplete(tag)
        case parseComplete:
        r.conn.bufferParseComplete()
    case bindComplete:
        r.conn.bufferBindComplete()
    case closeComplete:
        r.conn.bufferCloseComplete()
    case readyForQuery:
        r.conn.bufferReadyForQuery(byte(t))
        // The error is saved on conn.err.
        _ /* err */= r.conn.Flush(r.pos)
        ...
    }
    ...
}

客户端依据承受 SQL 申请的后果。

func(cn *conn) simpleQuery(q string)(res *rows, err error){b := cn.writeBuf('Q')
    b.string(q)
    cn.send(b)

    for{t, r := cn.recv1()
        switch t {
        case'C','I':
            ...
        case'Z':
            ...
        case'E':
            ...
        case'D':
            ...
        case'T':
            ...
        }
    }
}

Part 2 – 扩大查问

客户端发送扩大查问申请，顺次发送 Parse (F:‘P’), Bind (F:‘B’), Describe (F:‘D’), Execute (F:‘E’), Sync(F:‘S’) 音讯。

func(cn *conn) query(query string, args []driver.Value)(_ *rows, err error){
    ...
    if cn.binaryParameters {cn.sendBinaryModeQuery(query, args)
        cn.readParseResponse()
        cn.readBindResponse()
        rows :=&rows{cn: cn}
        rows.rowsHeader = cn.readPortalDescribeResponse()
        cn.postExecuteWorkaround()
        return rows,nil
    }
    ...
}

func(cn *conn) sendBinaryModeQuery(query string, args []driver.Value){b := cn.writeBuf('P')
    b.byte(0)//unnamed statement
    b.string(query)
    b.int16(0)

    b.next('B')
    b.int16(0)//unnamed portal and statement
    cn.sendBinaryParameters(b, args)
    b.bytes(colFmtDataAllText)

    b.next('D')
    b.byte('P')
    b.byte(0)//unnamed portal

    b.next('E')
    b.byte(0)
    b.int32(0)

    b.next('S')
    cn.send(b)
}

服务端解决扩大查问申请。

服务端发送回应音讯，ParseComplete (B:‘1’), BindComplete (B:‘2’), ParameterDescription(B:‘t’), CommandComplete (B:‘C’), CloseComplete (B:‘3’), ReadyForQuery (B:‘Z’)。

func(r *commandResult) Close(ctx context.Context, t sql.TransactionStatusIndicator){
    ...
    switch r.typ {
    case commandComplete:
        tag := cookTag(r.cmdCompleteTag, r.conn.writerState.tagBuf[:0], r.stmtType, r.rowsAffected,
        )
        r.conn.bufferCommandComplete(tag)
        case parseComplete:
        r.conn.bufferParseComplete()
    case bindComplete:
        r.conn.bufferBindComplete()
    case closeComplete:
        r.conn.bufferCloseComplete()
    case readyForQuery:
        r.conn.bufferReadyForQuery(byte(t))
        // The error is saved on conn.err.
        _ /* err */= r.conn.Flush(r.pos)
        ...
    }
    ...
}