From 8c5184890db250f755ff2a41b76839ce8419f930 Mon Sep 17 00:00:00 2001 From: t_max <1172915550@qq.com> Date: Fri, 17 Nov 2023 10:42:11 +0800 Subject: [PATCH] enh: support taos_write_raw_block_with_reqid and taos_write_raw_block_with_fields_with_reqid --- wrapper/block.go | 14 ++ wrapper/block_test.go | 371 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 385 insertions(+) diff --git a/wrapper/block.go b/wrapper/block.go index 30ef5a6..eacf894 100644 --- a/wrapper/block.go +++ b/wrapper/block.go @@ -33,3 +33,17 @@ func TaosWriteRawBlockWithFields(conn unsafe.Pointer, numOfRows int, pData unsaf defer C.free(unsafe.Pointer(cStr)) return int(C.taos_write_raw_block_with_fields(conn, (C.int)(numOfRows), (*C.char)(pData), cStr, (*C.struct_taosField)(fields), (C.int)(numFields))) } + +// DLL_EXPORT int taos_write_raw_block_with_reqid(TAOS *taos, int numOfRows, char *pData, const char *tbname, int64_t reqid); +func TaosWriteRawBlockWithReqID(conn unsafe.Pointer, numOfRows int, pData unsafe.Pointer, tableName string, reqID int64) int { + cStr := C.CString(tableName) + defer C.free(unsafe.Pointer(cStr)) + return int(C.taos_write_raw_block_with_reqid(conn, (C.int)(numOfRows), (*C.char)(pData), cStr, (C.int64_t)(reqID))) +} + +// DLL_EXPORT int taos_write_raw_block_with_fields_with_reqid(TAOS *taos, int rows, char *pData, const char *tbname,TAOS_FIELD *fields, int numFields, int64_t reqid); +func TaosWriteRawBlockWithFieldsWithReqID(conn unsafe.Pointer, numOfRows int, pData unsafe.Pointer, tableName string, fields unsafe.Pointer, numFields int, reqID int64) int { + cStr := C.CString(tableName) + defer C.free(unsafe.Pointer(cStr)) + return int(C.taos_write_raw_block_with_fields_with_reqid(conn, (C.int)(numOfRows), (*C.char)(pData), cStr, (*C.struct_taosField)(fields), (C.int)(numFields), (C.int64_t)(reqID))) +} diff --git a/wrapper/block_test.go b/wrapper/block_test.go index 12d912f..22f6239 100644 --- a/wrapper/block_test.go +++ b/wrapper/block_test.go @@ -551,3 +551,374 @@ func TestTaosWriteRawBlockWithFields(t *testing.T) { assert.Nil(t, row2[i]) } } + +// @author: xftan +// @date: 2023/11/17 9:39 +// @description: test write raw block with reqid +func TestTaosWriteRawBlockWithReqID(t *testing.T) { + conn, err := TaosConnect("", "root", "taosdata", "", 0) + if err != nil { + t.Error(err) + return + } + + defer TaosClose(conn) + res := TaosQuery(conn, "drop database if exists test_write_block_raw_with_reqid") + code := TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + defer func() { + res = TaosQuery(conn, "drop database if exists test_write_block_raw_with_reqid") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + }() + res = TaosQuery(conn, "create database test_write_block_raw_with_reqid") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + res = TaosQuery(conn, "create table if not exists test_write_block_raw_with_reqid.all_type (ts timestamp,"+ + "c1 bool,"+ + "c2 tinyint,"+ + "c3 smallint,"+ + "c4 int,"+ + "c5 bigint,"+ + "c6 tinyint unsigned,"+ + "c7 smallint unsigned,"+ + "c8 int unsigned,"+ + "c9 bigint unsigned,"+ + "c10 float,"+ + "c11 double,"+ + "c12 binary(20),"+ + "c13 nchar(20)"+ + ") tags (info json)") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + now := time.Now() + after1s := now.Add(time.Second) + sql := fmt.Sprintf("insert into test_write_block_raw_with_reqid.t0 using test_write_block_raw_with_reqid.all_type tags('{\"a\":1}') values('%s',1,1,1,1,1,1,1,1,1,1,1,'test_binary','test_nchar')('%s',null,null,null,null,null,null,null,null,null,null,null,null,null)", now.Format(time.RFC3339Nano), after1s.Format(time.RFC3339Nano)) + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "create table test_write_block_raw_with_reqid.t1 using test_write_block_raw_with_reqid.all_type tags('{\"a\":2}')" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "use test_write_block_raw_with_reqid" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "select * from test_write_block_raw_with_reqid.t0" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + for { + blockSize, errCode, block := TaosFetchRawBlock(res) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(res) + err := errors.NewError(errCode, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + if blockSize == 0 { + break + } + + errCode = TaosWriteRawBlockWithReqID(conn, blockSize, block, "t1", 1) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(nil) + err := errors.NewError(errCode, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + } + TaosFreeResult(res) + + sql = "select * from test_write_block_raw_with_reqid.t1" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + fileCount := TaosNumFields(res) + rh, err := ReadColumn(res, fileCount) + if err != nil { + t.Error(err) + return + } + precision := TaosResultPrecision(res) + var data [][]driver.Value + for { + blockSize, errCode, block := TaosFetchRawBlock(res) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(res) + err := errors.NewError(code, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + if blockSize == 0 { + break + } + d := parser.ReadBlock(block, blockSize, rh.ColTypes, precision) + data = append(data, d...) + } + TaosFreeResult(res) + + assert.Equal(t, 2, len(data)) + row1 := data[0] + assert.Equal(t, now.UnixNano()/1e6, row1[0].(time.Time).UnixNano()/1e6) + assert.Equal(t, true, row1[1].(bool)) + assert.Equal(t, int8(1), row1[2].(int8)) + assert.Equal(t, int16(1), row1[3].(int16)) + assert.Equal(t, int32(1), row1[4].(int32)) + assert.Equal(t, int64(1), row1[5].(int64)) + assert.Equal(t, uint8(1), row1[6].(uint8)) + assert.Equal(t, uint16(1), row1[7].(uint16)) + assert.Equal(t, uint32(1), row1[8].(uint32)) + assert.Equal(t, uint64(1), row1[9].(uint64)) + assert.Equal(t, float32(1), row1[10].(float32)) + assert.Equal(t, float64(1), row1[11].(float64)) + assert.Equal(t, "test_binary", row1[12].(string)) + assert.Equal(t, "test_nchar", row1[13].(string)) + row2 := data[1] + assert.Equal(t, after1s.UnixNano()/1e6, row2[0].(time.Time).UnixNano()/1e6) + for i := 1; i < 14; i++ { + assert.Nil(t, row2[i]) + } +} + +// @author: xftan +// @date: 2023/11/17 9:37 +// @description: test write raw block with fields and reqid +func TestTaosWriteRawBlockWithFieldsWithReqID(t *testing.T) { + conn, err := TaosConnect("", "root", "taosdata", "", 0) + if err != nil { + t.Error(err) + return + } + + defer TaosClose(conn) + res := TaosQuery(conn, "drop database if exists test_write_block_raw_fields_with_reqid") + code := TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + defer func() { + res = TaosQuery(conn, "drop database if exists test_write_block_raw_fields_with_reqid") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + }() + res = TaosQuery(conn, "create database test_write_block_raw_fields_with_reqid") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + res = TaosQuery(conn, "create table if not exists test_write_block_raw_fields_with_reqid.all_type (ts timestamp,"+ + "c1 bool,"+ + "c2 tinyint,"+ + "c3 smallint,"+ + "c4 int,"+ + "c5 bigint,"+ + "c6 tinyint unsigned,"+ + "c7 smallint unsigned,"+ + "c8 int unsigned,"+ + "c9 bigint unsigned,"+ + "c10 float,"+ + "c11 double,"+ + "c12 binary(20),"+ + "c13 nchar(20)"+ + ") tags (info json)") + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + now := time.Now() + after1s := now.Add(time.Second) + sql := fmt.Sprintf("insert into test_write_block_raw_fields_with_reqid.t0 using test_write_block_raw_fields_with_reqid.all_type tags('{\"a\":1}') values('%s',1,1,1,1,1,1,1,1,1,1,1,'test_binary','test_nchar')('%s',null,null,null,null,null,null,null,null,null,null,null,null,null)", now.Format(time.RFC3339Nano), after1s.Format(time.RFC3339Nano)) + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "create table test_write_block_raw_fields_with_reqid.t1 using test_write_block_raw_fields_with_reqid.all_type tags('{\"a\":2}')" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "use test_write_block_raw_fields_with_reqid" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + TaosFreeResult(res) + + sql = "select ts,c1 from test_write_block_raw_fields_with_reqid.t0" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + for { + blockSize, errCode, block := TaosFetchRawBlock(res) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(res) + err := errors.NewError(errCode, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + if blockSize == 0 { + break + } + fieldsCount := TaosNumFields(res) + fields := TaosFetchFields(res) + + errCode = TaosWriteRawBlockWithFieldsWithReqID(conn, blockSize, block, "t1", fields, fieldsCount, 1) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(nil) + err := errors.NewError(errCode, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + } + TaosFreeResult(res) + + sql = "select * from test_write_block_raw_fields_with_reqid.t1" + res = TaosQuery(conn, sql) + code = TaosError(res) + if code != 0 { + errStr := TaosErrorStr(res) + TaosFreeResult(res) + t.Error(errors.NewError(code, errStr)) + return + } + fileCount := TaosNumFields(res) + rh, err := ReadColumn(res, fileCount) + if err != nil { + t.Error(err) + return + } + precision := TaosResultPrecision(res) + var data [][]driver.Value + for { + blockSize, errCode, block := TaosFetchRawBlock(res) + if errCode != int(errors.SUCCESS) { + errStr := TaosErrorStr(res) + err := errors.NewError(code, errStr) + t.Error(err) + TaosFreeResult(res) + return + } + if blockSize == 0 { + break + } + d := parser.ReadBlock(block, blockSize, rh.ColTypes, precision) + data = append(data, d...) + } + TaosFreeResult(res) + + assert.Equal(t, 2, len(data)) + row1 := data[0] + assert.Equal(t, now.UnixNano()/1e6, row1[0].(time.Time).UnixNano()/1e6) + assert.Equal(t, true, row1[1].(bool)) + for i := 2; i < 14; i++ { + assert.Nil(t, row1[i]) + } + row2 := data[1] + assert.Equal(t, after1s.UnixNano()/1e6, row2[0].(time.Time).UnixNano()/1e6) + for i := 1; i < 14; i++ { + assert.Nil(t, row2[i]) + } +}