This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iotdb-client-go.git
The following commit(s) were added to refs/heads/main by this push:
new 013e3d3 Support OBJECT data type in tablet write (#175)
013e3d3 is described below
commit 013e3d3021424c286863c199e87c2fa9ae7f1020
Author: shuwenwei <[email protected]>
AuthorDate: Fri Aug 21 08:54:02 2026 +0800
Support OBJECT data type in tablet write (#175)
---
client/column_decoder.go | 2 +-
client/protocol.go | 3 +
client/rpcdataset.go | 12 ++
client/rpcdataset_test.go | 90 +++++++++++
client/sessiondataset.go | 15 ++
client/tablet.go | 55 ++++++-
client/tablet_test.go | 116 +++++++++++++++
client/utils.go | 24 +++
client/utils_test.go | 32 ++++
common/common.go | 205 ++++++++++++++++++++------
database/column/column.go | 4 +
database/column/{column.go => object.go} | 80 ++++------
database/column/{column.go => object_test.go} | 72 ++-------
test/e2e/e2e_table_test.go | 126 ++++++++++++++++
14 files changed, 679 insertions(+), 157 deletions(-)
diff --git a/client/column_decoder.go b/client/column_decoder.go
index 0898c4f..cd20b79 100644
--- a/client/column_decoder.go
+++ b/client/column_decoder.go
@@ -244,7 +244,7 @@ func (decoder *BinaryArrayColumnDecoder) ReadColumn(reader
*bytes.Reader, dataTy
// | int32 | bytes |
// +---------------+-------+
- if TEXT != dataType {
+ if TEXT != dataType && STRING != dataType && BLOB != dataType && OBJECT
!= dataType {
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
diff --git a/client/protocol.go b/client/protocol.go
index faedfad..111e642 100644
--- a/client/protocol.go
+++ b/client/protocol.go
@@ -39,6 +39,7 @@ const (
DATE TSDataType = 9
BLOB TSDataType = 10
STRING TSDataType = 11
+ OBJECT TSDataType = 12
)
var tsTypeMap = map[string]TSDataType{
@@ -52,6 +53,7 @@ var tsTypeMap = map[string]TSDataType{
"DATE": DATE,
"BLOB": BLOB,
"STRING": STRING,
+ "OBJECT": OBJECT,
}
var byteToTsDataType = map[byte]TSDataType{
@@ -65,6 +67,7 @@ var byteToTsDataType = map[byte]TSDataType{
9: DATE,
10: BLOB,
11: STRING,
+ 12: OBJECT,
}
func GetDataTypeByStr(name string) (TSDataType, error) {
diff --git a/client/rpcdataset.go b/client/rpcdataset.go
index d41a765..cba4d4c 100644
--- a/client/rpcdataset.go
+++ b/client/rpcdataset.go
@@ -548,6 +548,12 @@ func (s *IoTDBRpcDataSet)
getObjectByTsBlockIndex(tsBlockColumnIndex int32) (int
} else {
return binary.GetValues(), nil
}
+ case OBJECT:
+ if binary, err :=
s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBinary(s.tsBlockIndex); err !=
nil {
+ return nil, err
+ } else {
+ return objectBytesToString(binary.GetValues())
+ }
case DATE:
if value, err :=
s.curTsBlock.GetColumn(tsBlockColumnIndex).GetInt(s.tsBlockIndex); err != nil {
return nil, err
@@ -645,6 +651,12 @@ func (s *IoTDBRpcDataSet)
getStringByTsBlockColumnIndexAndDataType(index int32,
} else {
return bytesToHexString(v.values), nil
}
+ case OBJECT:
+ if v, err :=
s.curTsBlock.GetColumn(index).GetBinary(s.tsBlockIndex); err != nil {
+ return "", err
+ } else {
+ return objectBytesToString(v.values)
+ }
case DATE:
v, err := s.curTsBlock.GetColumn(index).GetInt(s.tsBlockIndex)
if err != nil {
diff --git a/client/rpcdataset_test.go b/client/rpcdataset_test.go
new file mode 100644
index 0000000..d9a5450
--- /dev/null
+++ b/client/rpcdataset_test.go
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package client
+
+import (
+ "encoding/binary"
+ "strings"
+ "testing"
+)
+
+func newObjectSessionDataSet(t *testing.T, value []byte) *SessionDataSet {
+ t.Helper()
+
+ column, err := NewBinaryColumn(0, 1, nil, []*Binary{NewBinary(value)})
+ if err != nil {
+ t.Fatalf("NewBinaryColumn() error = %v", err)
+ }
+ block, err := NewTsBlock(1, nil, column)
+ if err != nil {
+ t.Fatalf("NewTsBlock() error = %v", err)
+ }
+
+ return &SessionDataSet{ioTDBRpcDataSet: &IoTDBRpcDataSet{
+ columnNameList: []string{"file"},
+ columnTypeList: []string{"OBJECT"},
+ columnName2TsBlockColumnIndexMap: map[string]int32{"file": 0},
+ columnIndex2TsBlockColumnIndexList: []int32{0},
+ dataTypeForTsBlockColumn: []TSDataType{OBJECT},
+ queryResult: [][]byte{{1}},
+ queryResultSize: 1,
+ curTsBlock: block,
+ tsBlockSize: 1,
+ tsBlockIndex: 0,
+ }}
+}
+
+func TestSessionDataSet_OBJECTGetters(t *testing.T) {
+ value := make([]byte, 8+len("internal/path/1.bin"))
+ binary.BigEndian.PutUint64(value[:8], 1024)
+ copy(value[8:], "internal/path/1.bin")
+ dataSet := newObjectSessionDataSet(t, value)
+
+ object, err := dataSet.GetObject("file")
+ if err != nil {
+ t.Fatalf("GetObject() error = %v", err)
+ }
+ if object != "(Object) 1.00 KB" {
+ t.Errorf("GetObject() = %#v, want %q", object, "(Object) 1.00
KB")
+ }
+
+ object, err = dataSet.GetObjectByIndex(1)
+ if err != nil {
+ t.Fatalf("GetObjectByIndex() error = %v", err)
+ }
+ if object != "(Object) 1.00 KB" {
+ t.Errorf("GetObjectByIndex() = %#v, want %q", object, "(Object)
1.00 KB")
+ }
+
+ stringValue, err := dataSet.GetString("file")
+ if err != nil {
+ t.Fatalf("GetString() error = %v", err)
+ }
+ if stringValue != "(Object) 1.00 KB" {
+ t.Errorf("GetString() = %q, want %q", stringValue, "(Object)
1.00 KB")
+ }
+
+ if _, err := dataSet.GetBlob("file"); err == nil ||
!strings.Contains(err.Error(), "OBJECT") {
+ t.Fatalf("GetBlob() error = %v, want an OBJECT type error", err)
+ }
+ if _, err := dataSet.GetBlobByIndex(1); err == nil ||
!strings.Contains(err.Error(), "OBJECT") {
+ t.Fatalf("GetBlobByIndex() error = %v, want an OBJECT type
error", err)
+ }
+}
diff --git a/client/sessiondataset.go b/client/sessiondataset.go
index 66ffab1..58a9da3 100644
--- a/client/sessiondataset.go
+++ b/client/sessiondataset.go
@@ -1,6 +1,7 @@
package client
import (
+ "errors"
"time"
"github.com/apache/iotdb-client-go/v2/rpc"
@@ -107,10 +108,24 @@ func (s *SessionDataSet) GetDate(columnName string)
(time.Time, error) {
}
func (s *SessionDataSet) GetBlobByIndex(columnIndex int32) (*Binary, error) {
+ dataType, err := s.ioTDBRpcDataSet.getDataTypeByIndex(columnIndex)
+ if err != nil {
+ return nil, err
+ }
+ if dataType == OBJECT {
+ return nil, errors.New("OBJECT type does not support GetBlob")
+ }
return s.ioTDBRpcDataSet.getBinaryByIndex(columnIndex)
}
func (s *SessionDataSet) GetBlob(columnName string) (*Binary, error) {
+ dataType, err := s.ioTDBRpcDataSet.getDataType(columnName)
+ if err != nil {
+ return nil, err
+ }
+ if dataType == OBJECT {
+ return nil, errors.New("OBJECT type does not support GetBlob")
+ }
return s.ioTDBRpcDataSet.getBinary(columnName)
}
diff --git a/client/tablet.go b/client/tablet.go
index d694d6f..b919be9 100644
--- a/client/tablet.go
+++ b/client/tablet.go
@@ -74,7 +74,7 @@ func (t *Tablet) Swap(i, j int) {
case DOUBLE:
sortedSlice := t.values[index].([]float64)
sortedSlice[i], sortedSlice[j] = sortedSlice[j],
sortedSlice[i]
- case TEXT, BLOB, STRING:
+ case TEXT, BLOB, STRING, OBJECT:
sortedSlice := t.values[index].([][]byte)
sortedSlice[i], sortedSlice[j] = sortedSlice[j],
sortedSlice[i]
}
@@ -213,6 +213,8 @@ func (t *Tablet) SetValueAt(value interface{}, columnIndex,
rowIndex int) error
default:
return fmt.Errorf("illegal argument value %v %v",
value, reflect.TypeOf(value))
}
+ case OBJECT:
+ return fmt.Errorf("OBJECT values must be set with
SetObjectValueAt")
case DATE:
values := t.values[columnIndex].([]int32)
switch v := value.(type) {
@@ -226,6 +228,51 @@ func (t *Tablet) SetValueAt(value interface{},
columnIndex, rowIndex int) error
return fmt.Errorf("illegal argument value %v %v",
value, reflect.TypeOf(value))
}
}
+ t.unmarkNullValueAt(columnIndex, rowIndex)
+ return nil
+}
+
+func (t *Tablet) unmarkNullValueAt(columnIndex, rowIndex int) {
+ if t.bitMaps != nil && t.bitMaps[columnIndex] != nil {
+ t.bitMaps[columnIndex].UnMark(rowIndex)
+ }
+}
+
+// SetObjectValueAt writes a segment of an OBJECT column value. An OBJECT
value can be
+// written in multiple segments so that a large object does not need to be
fully loaded
+// into memory: each segment is wrapped into a 9-byte header (1 byte isEOF
flag followed
+// by an 8-byte big-endian offset) and then the raw content, consistent with
the Java
+// Tablet.addValue(rowIndex, columnIndex, isEOF, offset, content). Segments of
the same
+// object must be written in order with ascending offsets, and the last
segment must set
+// isEOF to true.
+//
+// Parameters:
+// - isEOF: Whether this segment is the last one of the object.
+// - offset: The offset of this segment within the whole object.
+// - content: The raw bytes of this segment.
+// - columnIndex: The column index of the OBJECT column.
+// - rowIndex: The row index to write the segment into.
+//
+// Returns:
+// - err: An error if the column/row index is invalid or the column is not
of type OBJECT.
+func (t *Tablet) SetObjectValueAt(isEOF bool, offset int64, content []byte,
columnIndex, rowIndex int) error {
+ if columnIndex < 0 || columnIndex >= len(t.measurementSchemas) {
+ return fmt.Errorf("illegal argument columnIndex %d",
columnIndex)
+ }
+ if rowIndex < 0 || rowIndex >= t.maxRowNumber {
+ return fmt.Errorf("illegal argument rowIndex %d", rowIndex)
+ }
+ if t.measurementSchemas[columnIndex].DataType != OBJECT {
+ return fmt.Errorf("column %d must be of type OBJECT",
columnIndex)
+ }
+ value := make([]byte, len(content)+9)
+ if isEOF {
+ value[0] = 1
+ }
+ binary.BigEndian.PutUint64(value[1:9], uint64(offset))
+ copy(value[9:], content)
+ t.values[columnIndex].([][]byte)[rowIndex] = value
+ t.unmarkNullValueAt(columnIndex, rowIndex)
return nil
}
@@ -260,7 +307,7 @@ func (t *Tablet) GetValueAt(columnIndex, rowIndex int)
(interface{}, error) {
return t.values[columnIndex].([]float64)[rowIndex], nil
case TEXT, STRING:
return string(t.values[columnIndex].([][]byte)[rowIndex]), nil
- case BLOB:
+ case BLOB, OBJECT:
return t.values[columnIndex].([][]byte)[rowIndex], nil
case DATE:
return Int32ToDate(t.values[columnIndex].([]int32)[rowIndex])
@@ -313,7 +360,7 @@ func (t *Tablet) getValuesBytes() ([]byte, error) {
binary.Write(buff, binary.BigEndian,
t.values[i].([]float32)[0:t.RowSize])
case DOUBLE:
binary.Write(buff, binary.BigEndian,
t.values[i].([]float64)[0:t.RowSize])
- case TEXT, STRING, BLOB:
+ case TEXT, STRING, BLOB, OBJECT:
for _, s := range t.values[i].([][]byte)[0:t.RowSize] {
binary.Write(buff, binary.BigEndian,
int32(len(s)))
binary.Write(buff, binary.BigEndian, s)
@@ -365,7 +412,7 @@ func NewTablet(insertTargetName string, measurementSchemas
[]*MeasurementSchema,
tablet.values[i] = make([]float32, maxRowNumber)
case DOUBLE:
tablet.values[i] = make([]float64, maxRowNumber)
- case TEXT, STRING, BLOB:
+ case TEXT, STRING, BLOB, OBJECT:
tablet.values[i] = make([][]byte, maxRowNumber)
default:
return nil, fmt.Errorf("illegal datatype %v",
schema.DataType)
diff --git a/client/tablet_test.go b/client/tablet_test.go
index d6f701d..be1f480 100644
--- a/client/tablet_test.go
+++ b/client/tablet_test.go
@@ -676,3 +676,119 @@ func TestTablet_Sort(t *testing.T) {
})
}
}
+
+func TestTablet_OBJECT(t *testing.T) {
+ tablet, err := NewRelationalTablet("t1", []*MeasurementSchema{
+ {Measurement: "tag1", DataType: STRING},
+ {Measurement: "obj", DataType: OBJECT},
+ }, []ColumnCategory{TAG, FIELD}, 4)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if got := tablet.getColumnCategories(); !reflect.DeepEqual(got,
[]int8{0, 1}) {
+ t.Errorf("getColumnCategories() = %v, want [0 1]", got)
+ }
+
+ if got := tablet.getDataTypes(); !reflect.DeepEqual(got, []int32{11,
12}) {
+ t.Errorf("getDataTypes() = %v, want [11 12]", got)
+ }
+
+ objVal := []byte{0x01, 0x02, 0x03}
+ if err := tablet.SetValueAt(objVal, 1, 0); err == nil {
+ t.Fatal("SetValueAt([]byte) for OBJECT: want error, got nil")
+ }
+ if err := tablet.SetObjectValueAt(true, 0, objVal, 1, 0); err != nil {
+ t.Fatalf("SetObjectValueAt([]byte) error = %v", err)
+ }
+ tablet.SetTimestamp(1608268702780, 0)
+ tablet.RowSize++
+
+ if err := tablet.SetValueAt("hello", 1, 1); err == nil {
+ t.Fatal("SetValueAt(string) for OBJECT: want error, got nil")
+ }
+ if err := tablet.SetObjectValueAt(true, 0, []byte("hello"), 1, 1); err
!= nil {
+ t.Fatalf("SetObjectValueAt(string bytes) error = %v", err)
+ }
+ tablet.SetTimestamp(1608268702781, 1)
+ tablet.RowSize++
+
+ wantObject1 := append([]byte{1, 0, 0, 0, 0, 0, 0, 0, 0}, objVal...)
+ wantObject2 := append([]byte{1, 0, 0, 0, 0, 0, 0, 0, 0},
[]byte("hello")...)
+ if got, err := tablet.GetValueAt(1, 0); err != nil ||
!reflect.DeepEqual(got, wantObject1) {
+ t.Errorf("GetValueAt(1,0) = %v, %v; want %v, nil", got, err,
wantObject1)
+ }
+ if got, err := tablet.GetValueAt(1, 1); err != nil ||
!reflect.DeepEqual(got, wantObject2) {
+ t.Errorf("GetValueAt(1,1) = %v, %v; want %v, nil", got, err,
wantObject2)
+ }
+
+ valuesBytes, err := tablet.getValuesBytes()
+ if err != nil {
+ t.Fatal(err)
+ }
+ wantValues := []byte{
+ 0, 0, 0, 0,
+ 0, 0, 0, 0,
+ 0, 0, 0, byte(len(wantObject1)),
+ }
+ wantValues = append(wantValues, wantObject1...)
+ wantValues = append(wantValues, 0, 0, 0, byte(len(wantObject2)))
+ wantValues = append(wantValues, wantObject2...)
+ if !reflect.DeepEqual(valuesBytes, wantValues) {
+ t.Errorf("getValuesBytes() = %v, want %v", valuesBytes,
wantValues)
+ }
+}
+
+func TestTablet_SetObjectValueAt(t *testing.T) {
+ tablet, err := NewRelationalTablet("t1", []*MeasurementSchema{
+ {Measurement: "tag1", DataType: STRING},
+ {Measurement: "obj", DataType: OBJECT},
+ }, []ColumnCategory{TAG, FIELD}, 4)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ tablet.SetTimestamp(1, 0)
+ if err := tablet.SetObjectValueAt(false, 0, []byte{0x11, 0x22}, 1, 0);
err != nil {
+ t.Fatalf("SetObjectValueAt(segment) error = %v", err)
+ }
+ tablet.RowSize++
+
+ tablet.SetTimestamp(1, 1)
+ if err := tablet.SetObjectValueAt(true, 512, []byte{0x33}, 1, 1); err
!= nil {
+ t.Fatalf("SetObjectValueAt(last segment) error = %v", err)
+ }
+ tablet.RowSize++
+
+ wantSegment1 := []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22}
+ wantSegment2 := []byte{1, 0, 0, 0, 0, 0, 0, 2, 0, 0x33}
+ if got, err := tablet.GetValueAt(1, 0); err != nil ||
!reflect.DeepEqual(got, wantSegment1) {
+ t.Errorf("GetValueAt(1,0) = %v, %v; want %v, nil", got, err,
wantSegment1)
+ }
+ if got, err := tablet.GetValueAt(1, 1); err != nil ||
!reflect.DeepEqual(got, wantSegment2) {
+ t.Errorf("GetValueAt(1,1) = %v, %v; want %v, nil", got, err,
wantSegment2)
+ }
+
+ if err := tablet.SetValueAt(nil, 1, 2); err != nil {
+ t.Fatalf("SetValueAt(nil) error = %v", err)
+ }
+ if tablet.bitMaps == nil || tablet.bitMaps[1] == nil ||
!tablet.bitMaps[1].IsMarked(2) {
+ t.Fatal("SetValueAt(nil) did not mark the OBJECT cell as null")
+ }
+ if err := tablet.SetObjectValueAt(true, 0, []byte{0x44}, 1, 2); err !=
nil {
+ t.Fatalf("SetObjectValueAt() after null error = %v", err)
+ }
+ if tablet.bitMaps[1].IsMarked(2) {
+ t.Error("SetObjectValueAt() did not clear the previously marked
null bit")
+ }
+
+ if err := tablet.SetObjectValueAt(false, 0, []byte{0x01}, 0, 0); err ==
nil {
+ t.Error("SetObjectValueAt() on non-OBJECT column: want error,
got nil")
+ }
+ if err := tablet.SetObjectValueAt(false, 0, []byte{0x01}, 1, -1); err
== nil {
+ t.Error("SetObjectValueAt() with invalid rowIndex: want error,
got nil")
+ }
+ if err := tablet.SetObjectValueAt(false, 0, []byte{0x01}, -1, 0); err
== nil {
+ t.Error("SetObjectValueAt() with invalid columnIndex: want
error, got nil")
+ }
+}
diff --git a/client/utils.go b/client/utils.go
index 276cbe4..d6e8f93 100644
--- a/client/utils.go
+++ b/client/utils.go
@@ -206,6 +206,30 @@ func bytesToHexString(input []byte) string {
return hexString
}
+func objectBytesToString(input []byte) (string, error) {
+ const (
+ kilobyte = uint64(1024)
+ megabyte = kilobyte * 1024
+ gigabyte = megabyte * 1024
+ )
+
+ if len(input) < 8 {
+ return "", fmt.Errorf("invalid OBJECT value: expected at least
8 bytes, got %d", len(input))
+ }
+
+ size := binary.BigEndian.Uint64(input[:8])
+ switch {
+ case size < kilobyte:
+ return fmt.Sprintf("(Object) %d B", size), nil
+ case size < megabyte:
+ return fmt.Sprintf("(Object) %.2f KB",
float64(size)/float64(kilobyte)), nil
+ case size < gigabyte:
+ return fmt.Sprintf("(Object) %.2f MB",
float64(size)/float64(megabyte)), nil
+ default:
+ return fmt.Sprintf("(Object) %.2f GB",
float64(size)/float64(gigabyte)), nil
+ }
+}
+
func DateToInt32(localDate time.Time) (int32, error) {
if localDate.IsZero() {
return 0, errors.New("date expression is null or empty")
diff --git a/client/utils_test.go b/client/utils_test.go
index 95d8d2a..7623da1 100644
--- a/client/utils_test.go
+++ b/client/utils_test.go
@@ -20,6 +20,7 @@
package client
import (
+ "encoding/binary"
"testing"
"github.com/apache/iotdb-client-go/v2/common"
@@ -207,6 +208,37 @@ func Test_bytesToHexString(t *testing.T) {
}
}
+func Test_objectBytesToString(t *testing.T) {
+ tests := []struct {
+ name string
+ size uint64
+ want string
+ }{
+ {name: "bytes", size: 1023, want: "(Object) 1023 B"},
+ {name: "kilobytes", size: 1024, want: "(Object) 1.00 KB"},
+ {name: "megabytes", size: 1024 * 1024, want: "(Object) 1.00
MB"},
+ {name: "gigabytes", size: 1024 * 1024 * 1024, want: "(Object)
1.00 GB"},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ value := make([]byte, 8)
+ binary.BigEndian.PutUint64(value, tt.size)
+ got, err := objectBytesToString(value)
+ if err != nil {
+ t.Fatalf("objectBytesToString() error = %v",
err)
+ }
+ if got != tt.want {
+ t.Errorf("objectBytesToString() = %q, want %q",
got, tt.want)
+ }
+ })
+ }
+
+ if _, err := objectBytesToString(make([]byte, 7)); err == nil {
+ t.Fatal("objectBytesToString() with a short value: want error,
got nil")
+ }
+}
+
func Test_verifySuccess(t *testing.T) {
type args struct {
status *common.TSStatus
diff --git a/common/common.go b/common/common.go
index 74e03c6..18f4058 100644
--- a/common/common.go
+++ b/common/common.go
@@ -475,6 +475,10 @@ const (
TAggregationType_SKEWNESS TAggregationType = 38
TAggregationType_KURTOSIS TAggregationType = 39
TAggregationType_PERCENTILE TAggregationType = 40
+ TAggregationType_RATE TAggregationType = 41
+ TAggregationType_INCREASE TAggregationType = 42
+ TAggregationType_IRATE TAggregationType = 43
+ TAggregationType_DELTA TAggregationType = 44
)
var knownTAggregationTypeValues = []TAggregationType{
@@ -519,6 +523,10 @@ var knownTAggregationTypeValues = []TAggregationType{
TAggregationType_SKEWNESS,
TAggregationType_KURTOSIS,
TAggregationType_PERCENTILE,
+ TAggregationType_RATE,
+ TAggregationType_INCREASE,
+ TAggregationType_IRATE,
+ TAggregationType_DELTA,
}
func TAggregationTypeValues() iter.Seq[TAggregationType] {
@@ -574,6 +582,10 @@ func (p TAggregationType) String() string {
case TAggregationType_SKEWNESS: return "SKEWNESS"
case TAggregationType_KURTOSIS: return "KURTOSIS"
case TAggregationType_PERCENTILE: return "PERCENTILE"
+ case TAggregationType_RATE: return "RATE"
+ case TAggregationType_INCREASE: return "INCREASE"
+ case TAggregationType_IRATE: return "IRATE"
+ case TAggregationType_DELTA: return "DELTA"
}
return "<UNSET>"
}
@@ -621,6 +633,10 @@ func TAggregationTypeFromString(s string)
(TAggregationType, error) {
case "SKEWNESS": return TAggregationType_SKEWNESS, nil
case "KURTOSIS": return TAggregationType_KURTOSIS, nil
case "PERCENTILE": return TAggregationType_PERCENTILE, nil
+ case "RATE": return TAggregationType_RATE, nil
+ case "INCREASE": return TAggregationType_INCREASE, nil
+ case "IRATE": return TAggregationType_IRATE, nil
+ case "DELTA": return TAggregationType_DELTA, nil
}
return TAggregationType(0), fmt.Errorf("not a valid TAggregationType
string")
}
@@ -6310,6 +6326,7 @@ func (p *TSetThrottleQuotaReq) Validate() error {
// - PipeRemainingEventCountList
// - PipeRemainingTimeList
// - PipeDegradedStatusList
+// - PipeRecentFailureList
//
type TPipeHeartbeatResp struct {
PipeMetaList [][]byte `thrift:"pipeMetaList,1,required"
db:"pipeMetaList" json:"pipeMetaList"`
@@ -6317,6 +6334,7 @@ type TPipeHeartbeatResp struct {
PipeRemainingEventCountList []int64
`thrift:"pipeRemainingEventCountList,3" db:"pipeRemainingEventCountList"
json:"pipeRemainingEventCountList,omitempty"`
PipeRemainingTimeList []float64 `thrift:"pipeRemainingTimeList,4"
db:"pipeRemainingTimeList" json:"pipeRemainingTimeList,omitempty"`
PipeDegradedStatusList []int32 `thrift:"pipeDegradedStatusList,5"
db:"pipeDegradedStatusList" json:"pipeDegradedStatusList,omitempty"`
+ PipeRecentFailureList []map[string]int64
`thrift:"pipeRecentFailureList,6" db:"pipeRecentFailureList"
json:"pipeRecentFailureList,omitempty"`
}
func NewTPipeHeartbeatResp() *TPipeHeartbeatResp {
@@ -6357,6 +6375,13 @@ func (p *TPipeHeartbeatResp) GetPipeDegradedStatusList()
[]int32 {
return p.PipeDegradedStatusList
}
+var TPipeHeartbeatResp_PipeRecentFailureList_DEFAULT []map[string]int64
+
+
+func (p *TPipeHeartbeatResp) GetPipeRecentFailureList() []map[string]int64 {
+ return p.PipeRecentFailureList
+}
+
func (p *TPipeHeartbeatResp) IsSetPipeCompletedList() bool {
return p.PipeCompletedList != nil
}
@@ -6373,6 +6398,10 @@ func (p *TPipeHeartbeatResp)
IsSetPipeDegradedStatusList() bool {
return p.PipeDegradedStatusList != nil
}
+func (p *TPipeHeartbeatResp) IsSetPipeRecentFailureList() bool {
+ return p.PipeRecentFailureList != nil
+}
+
func (p *TPipeHeartbeatResp) Read(ctx context.Context, iprot thrift.TProtocol)
error {
if _, err := iprot.ReadStructBegin(ctx); err != nil {
return thrift.PrependError(fmt.Sprintf("%T read error: ", p),
err)
@@ -6440,6 +6469,16 @@ func (p *TPipeHeartbeatResp) Read(ctx context.Context,
iprot thrift.TProtocol) e
return err
}
}
+ case 6:
+ if fieldTypeId == thrift.LIST {
+ if err := p.ReadField6(ctx, iprot); err != nil {
+ return err
+ }
+ } else {
+ if err := iprot.Skip(ctx, fieldTypeId); err !=
nil {
+ return err
+ }
+ }
default:
if err := iprot.Skip(ctx, fieldTypeId); err != nil {
return err
@@ -6568,6 +6607,46 @@ func (p *TPipeHeartbeatResp) ReadField5(ctx
context.Context, iprot thrift.TProto
return nil
}
+func (p *TPipeHeartbeatResp) ReadField6(ctx context.Context, iprot
thrift.TProtocol) error {
+ _, size, err := iprot.ReadListBegin(ctx)
+ if err != nil {
+ return thrift.PrependError("error reading list begin: ", err)
+ }
+ tSlice := make([]map[string]int64, 0, size)
+ p.PipeRecentFailureList = tSlice
+ for i := 0; i < size; i++ {
+ _, _, size, err := iprot.ReadMapBegin(ctx)
+ if err != nil {
+ return thrift.PrependError("error reading map begin: ",
err)
+ }
+ tMap := make(map[string]int64, size)
+ _elem29 := tMap
+ for i := 0; i < size; i++ {
+ var _key30 string
+ if v, err := iprot.ReadString(ctx); err != nil {
+ return thrift.PrependError("error reading field
0: ", err)
+ } else {
+ _key30 = v
+ }
+ var _val31 int64
+ if v, err := iprot.ReadI64(ctx); err != nil {
+ return thrift.PrependError("error reading field
0: ", err)
+ } else {
+ _val31 = v
+ }
+ _elem29[_key30] = _val31
+ }
+ if err := iprot.ReadMapEnd(ctx); err != nil {
+ return thrift.PrependError("error reading map end: ",
err)
+ }
+ p.PipeRecentFailureList = append(p.PipeRecentFailureList,
_elem29)
+ }
+ if err := iprot.ReadListEnd(ctx); err != nil {
+ return thrift.PrependError("error reading list end: ", err)
+ }
+ return nil
+}
+
func (p *TPipeHeartbeatResp) Write(ctx context.Context, oprot
thrift.TProtocol) error {
if err := oprot.WriteStructBegin(ctx, "TPipeHeartbeatResp"); err != nil
{
return thrift.PrependError(fmt.Sprintf("%T write struct begin
error: ", p), err)
@@ -6578,6 +6657,7 @@ func (p *TPipeHeartbeatResp) Write(ctx context.Context,
oprot thrift.TProtocol)
if err := p.writeField3(ctx, oprot); err != nil { return err }
if err := p.writeField4(ctx, oprot); err != nil { return err }
if err := p.writeField5(ctx, oprot); err != nil { return err }
+ if err := p.writeField6(ctx, oprot); err != nil { return err }
}
if err := oprot.WriteFieldStop(ctx); err != nil {
return thrift.PrependError("write field stop error: ", err)
@@ -6701,6 +6781,40 @@ func (p *TPipeHeartbeatResp) writeField5(ctx
context.Context, oprot thrift.TProt
return err
}
+func (p *TPipeHeartbeatResp) writeField6(ctx context.Context, oprot
thrift.TProtocol) (err error) {
+ if p.IsSetPipeRecentFailureList() {
+ if err := oprot.WriteFieldBegin(ctx, "pipeRecentFailureList",
thrift.LIST, 6); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T write field
begin error 6:pipeRecentFailureList: ", p), err)
+ }
+ if err := oprot.WriteListBegin(ctx, thrift.MAP,
len(p.PipeRecentFailureList)); err != nil {
+ return thrift.PrependError("error writing list begin:
", err)
+ }
+ for _, v := range p.PipeRecentFailureList {
+ if err := oprot.WriteMapBegin(ctx, thrift.STRING,
thrift.I64, len(v)); err != nil {
+ return thrift.PrependError("error writing map
begin: ", err)
+ }
+ for k, v := range v {
+ if err := oprot.WriteString(ctx, string(k));
err != nil {
+ return
thrift.PrependError(fmt.Sprintf("%T. (0) field write error: ", p), err)
+ }
+ if err := oprot.WriteI64(ctx, int64(v)); err !=
nil {
+ return
thrift.PrependError(fmt.Sprintf("%T. (0) field write error: ", p), err)
+ }
+ }
+ if err := oprot.WriteMapEnd(ctx); err != nil {
+ return thrift.PrependError("error writing map
end: ", err)
+ }
+ }
+ if err := oprot.WriteListEnd(ctx); err != nil {
+ return thrift.PrependError("error writing list end: ",
err)
+ }
+ if err := oprot.WriteFieldEnd(ctx); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T write field
end error 6:pipeRecentFailureList: ", p), err)
+ }
+ }
+ return err
+}
+
func (p *TPipeHeartbeatResp) Equals(other *TPipeHeartbeatResp) bool {
if p == other {
return true
@@ -6709,28 +6823,37 @@ func (p *TPipeHeartbeatResp) Equals(other
*TPipeHeartbeatResp) bool {
}
if len(p.PipeMetaList) != len(other.PipeMetaList) { return false }
for i, _tgt := range p.PipeMetaList {
- _src29 := other.PipeMetaList[i]
- if bytes.Compare(_tgt, _src29) != 0 { return false }
+ _src32 := other.PipeMetaList[i]
+ if bytes.Compare(_tgt, _src32) != 0 { return false }
}
if len(p.PipeCompletedList) != len(other.PipeCompletedList) { return
false }
for i, _tgt := range p.PipeCompletedList {
- _src30 := other.PipeCompletedList[i]
- if _tgt != _src30 { return false }
+ _src33 := other.PipeCompletedList[i]
+ if _tgt != _src33 { return false }
}
if len(p.PipeRemainingEventCountList) !=
len(other.PipeRemainingEventCountList) { return false }
for i, _tgt := range p.PipeRemainingEventCountList {
- _src31 := other.PipeRemainingEventCountList[i]
- if _tgt != _src31 { return false }
+ _src34 := other.PipeRemainingEventCountList[i]
+ if _tgt != _src34 { return false }
}
if len(p.PipeRemainingTimeList) != len(other.PipeRemainingTimeList) {
return false }
for i, _tgt := range p.PipeRemainingTimeList {
- _src32 := other.PipeRemainingTimeList[i]
- if _tgt != _src32 { return false }
+ _src35 := other.PipeRemainingTimeList[i]
+ if _tgt != _src35 { return false }
}
if len(p.PipeDegradedStatusList) != len(other.PipeDegradedStatusList) {
return false }
for i, _tgt := range p.PipeDegradedStatusList {
- _src33 := other.PipeDegradedStatusList[i]
- if _tgt != _src33 { return false }
+ _src36 := other.PipeDegradedStatusList[i]
+ if _tgt != _src36 { return false }
+ }
+ if len(p.PipeRecentFailureList) != len(other.PipeRecentFailureList) {
return false }
+ for i, _tgt := range p.PipeRecentFailureList {
+ _src37 := other.PipeRecentFailureList[i]
+ if len(_tgt) != len(_src37) { return false }
+ for k, _tgt := range _tgt {
+ _src38 := _src37[k]
+ if _tgt != _src38 { return false }
+ }
}
return true
}
@@ -8324,11 +8447,11 @@ func (p *TTestConnectionResp) ReadField2(ctx
context.Context, iprot thrift.TProt
tSlice := make([]*TTestConnectionResult_, 0, size)
p.ResultList = tSlice
for i := 0; i < size; i++ {
- _elem34 := &TTestConnectionResult_{}
- if err := _elem34.Read(ctx, iprot); err != nil {
- return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem34), err)
+ _elem39 := &TTestConnectionResult_{}
+ if err := _elem39.Read(ctx, iprot); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem39), err)
}
- p.ResultList = append(p.ResultList, _elem34)
+ p.ResultList = append(p.ResultList, _elem39)
}
if err := iprot.ReadListEnd(ctx); err != nil {
return thrift.PrependError("error reading list end: ", err)
@@ -8396,8 +8519,8 @@ func (p *TTestConnectionResp) Equals(other
*TTestConnectionResp) bool {
if !p.Status.Equals(other.Status) { return false }
if len(p.ResultList) != len(other.ResultList) { return false }
for i, _tgt := range p.ResultList {
- _src35 := other.ResultList[i]
- if !_tgt.Equals(_src35) { return false }
+ _src40 := other.ResultList[i]
+ if !_tgt.Equals(_src40) { return false }
}
return true
}
@@ -8519,11 +8642,11 @@ func (p *TNodeLocations) ReadField1(ctx
context.Context, iprot thrift.TProtocol)
tSlice := make([]*TConfigNodeLocation, 0, size)
p.ConfigNodeLocations = tSlice
for i := 0; i < size; i++ {
- _elem36 := &TConfigNodeLocation{}
- if err := _elem36.Read(ctx, iprot); err != nil {
- return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem36), err)
+ _elem41 := &TConfigNodeLocation{}
+ if err := _elem41.Read(ctx, iprot); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem41), err)
}
- p.ConfigNodeLocations = append(p.ConfigNodeLocations, _elem36)
+ p.ConfigNodeLocations = append(p.ConfigNodeLocations, _elem41)
}
if err := iprot.ReadListEnd(ctx); err != nil {
return thrift.PrependError("error reading list end: ", err)
@@ -8539,11 +8662,11 @@ func (p *TNodeLocations) ReadField2(ctx
context.Context, iprot thrift.TProtocol)
tSlice := make([]*TDataNodeLocation, 0, size)
p.DataNodeLocations = tSlice
for i := 0; i < size; i++ {
- _elem37 := &TDataNodeLocation{}
- if err := _elem37.Read(ctx, iprot); err != nil {
- return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem37), err)
+ _elem42 := &TDataNodeLocation{}
+ if err := _elem42.Read(ctx, iprot); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem42), err)
}
- p.DataNodeLocations = append(p.DataNodeLocations, _elem37)
+ p.DataNodeLocations = append(p.DataNodeLocations, _elem42)
}
if err := iprot.ReadListEnd(ctx); err != nil {
return thrift.PrependError("error reading list end: ", err)
@@ -8622,13 +8745,13 @@ func (p *TNodeLocations) Equals(other *TNodeLocations)
bool {
}
if len(p.ConfigNodeLocations) != len(other.ConfigNodeLocations) {
return false }
for i, _tgt := range p.ConfigNodeLocations {
- _src38 := other.ConfigNodeLocations[i]
- if !_tgt.Equals(_src38) { return false }
+ _src43 := other.ConfigNodeLocations[i]
+ if !_tgt.Equals(_src43) { return false }
}
if len(p.DataNodeLocations) != len(other.DataNodeLocations) { return
false }
for i, _tgt := range p.DataNodeLocations {
- _src39 := other.DataNodeLocations[i]
- if !_tgt.Equals(_src39) { return false }
+ _src44 := other.DataNodeLocations[i]
+ if !_tgt.Equals(_src44) { return false }
}
return true
}
@@ -8765,11 +8888,11 @@ func (p *TExternalServiceListResp) ReadField2(ctx
context.Context, iprot thrift.
tSlice := make([]*TExternalServiceEntry, 0, size)
p.ExternalServiceInfos = tSlice
for i := 0; i < size; i++ {
- _elem40 := &TExternalServiceEntry{}
- if err := _elem40.Read(ctx, iprot); err != nil {
- return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem40), err)
+ _elem45 := &TExternalServiceEntry{}
+ if err := _elem45.Read(ctx, iprot); err != nil {
+ return thrift.PrependError(fmt.Sprintf("%T error
reading struct: ", _elem45), err)
}
- p.ExternalServiceInfos = append(p.ExternalServiceInfos, _elem40)
+ p.ExternalServiceInfos = append(p.ExternalServiceInfos, _elem45)
}
if err := iprot.ReadListEnd(ctx); err != nil {
return thrift.PrependError("error reading list end: ", err)
@@ -8837,8 +8960,8 @@ func (p *TExternalServiceListResp) Equals(other
*TExternalServiceListResp) bool
if !p.Status.Equals(other.Status) { return false }
if len(p.ExternalServiceInfos) != len(other.ExternalServiceInfos) {
return false }
for i, _tgt := range p.ExternalServiceInfos {
- _src41 := other.ExternalServiceInfos[i]
- if !_tgt.Equals(_src41) { return false }
+ _src46 := other.ExternalServiceInfos[i]
+ if !_tgt.Equals(_src46) { return false }
}
return true
}
@@ -9673,19 +9796,19 @@ func (p *TShowAppliedConfigurationsResp) ReadField2(ctx
context.Context, iprot t
tMap := make(map[string]string, size)
p.Data = tMap
for i := 0; i < size; i++ {
- var _key42 string
+ var _key47 string
if v, err := iprot.ReadString(ctx); err != nil {
return thrift.PrependError("error reading field 0: ",
err)
} else {
- _key42 = v
+ _key47 = v
}
- var _val43 string
+ var _val48 string
if v, err := iprot.ReadString(ctx); err != nil {
return thrift.PrependError("error reading field 0: ",
err)
} else {
- _val43 = v
+ _val48 = v
}
- p.Data[_key42] = _val43
+ p.Data[_key47] = _val48
}
if err := iprot.ReadMapEnd(ctx); err != nil {
return thrift.PrependError("error reading map end: ", err)
@@ -9758,8 +9881,8 @@ func (p *TShowAppliedConfigurationsResp) Equals(other
*TShowAppliedConfiguration
if !p.Status.Equals(other.Status) { return false }
if len(p.Data) != len(other.Data) { return false }
for k, _tgt := range p.Data {
- _src44 := other.Data[k]
- if _tgt != _src44 { return false }
+ _src49 := other.Data[k]
+ if _tgt != _src49 { return false }
}
return true
}
diff --git a/database/column/column.go b/database/column/column.go
index 802246b..9095617 100644
--- a/database/column/column.go
+++ b/database/column/column.go
@@ -77,6 +77,10 @@ func GenColumn(dataType string, name string) Interface {
return &String{
name: name,
}
+ case "OBJECT":
+ return &Object{
+ name: name,
+ }
}
return nil
}
diff --git a/database/column/column.go b/database/column/object.go
similarity index 51%
copy from database/column/column.go
copy to database/column/object.go
index 802246b..ebe8232 100644
--- a/database/column/column.go
+++ b/database/column/object.go
@@ -19,64 +19,38 @@
package column
-import (
- "github.com/apache/iotdb-client-go/v2/client"
-)
+import "github.com/apache/iotdb-client-go/v2/client"
-type Type string
+type Object struct {
+ name string
+}
-type Interface interface {
- Name() string
- Type() Type
- Row(stat *client.SessionDataSet, ptr bool) any
+func (o *Object) Name() string {
+ return o.name
}
-func GenColumn(dataType string, name string) Interface {
- switch dataType {
- case "BOOLEAN":
- return &Bool{
- name: name,
- }
- case "INT32":
- return &Int32{
- name: name,
- }
- case "INT64":
- {
- return &Int64{
- name: name,
- }
- }
- case "FLOAT":
- return &Float{
- name: name,
- }
- case "DOUBLE":
- return &Double{
- name: name,
- }
- case "TEXT":
- {
- return &String{
- name: name,
- }
- }
- case "TIMESTAMP":
- return &Timestamp{
- name: name,
- }
- case "DATE":
- return &Date{
- name: name,
- }
- case "BLOB":
- return &Blob{
- name: name,
+func (o *Object) Type() Type {
+ return "OBJECT"
+}
+
+func (o *Object) Row(stat *client.SessionDataSet, ptr bool) any {
+ if stat == nil {
+ if ptr {
+ return nil
}
- case "STRING":
- return &String{
- name: name,
+ return ""
+ }
+ value, err := stat.GetString(o.name)
+ if err != nil {
+ if ptr {
+ return nil
}
+ return ""
+ }
+ if ptr {
+ return &value
}
- return nil
+ return value
}
+
+var _ Interface = (*Object)(nil)
diff --git a/database/column/column.go b/database/column/object_test.go
similarity index 50%
copy from database/column/column.go
copy to database/column/object_test.go
index 802246b..8069af4 100644
--- a/database/column/column.go
+++ b/database/column/object_test.go
@@ -19,64 +19,20 @@
package column
-import (
- "github.com/apache/iotdb-client-go/v2/client"
-)
+import "testing"
-type Type string
-
-type Interface interface {
- Name() string
- Type() Type
- Row(stat *client.SessionDataSet, ptr bool) any
-}
-
-func GenColumn(dataType string, name string) Interface {
- switch dataType {
- case "BOOLEAN":
- return &Bool{
- name: name,
- }
- case "INT32":
- return &Int32{
- name: name,
- }
- case "INT64":
- {
- return &Int64{
- name: name,
- }
- }
- case "FLOAT":
- return &Float{
- name: name,
- }
- case "DOUBLE":
- return &Double{
- name: name,
- }
- case "TEXT":
- {
- return &String{
- name: name,
- }
- }
- case "TIMESTAMP":
- return &Timestamp{
- name: name,
- }
- case "DATE":
- return &Date{
- name: name,
- }
- case "BLOB":
- return &Blob{
- name: name,
- }
- case "STRING":
- return &String{
- name: name,
- }
+func TestGenColumn_OBJECT(t *testing.T) {
+ column := GenColumn("OBJECT", "file")
+ if column == nil {
+ t.Fatal("GenColumn() returned nil for OBJECT")
+ }
+ if _, ok := column.(*Object); !ok {
+ t.Fatalf("GenColumn() returned %T for OBJECT, want *Object",
column)
+ }
+ if column.Name() != "file" {
+ t.Errorf("Name() = %q, want %q", column.Name(), "file")
+ }
+ if column.Type() != "OBJECT" {
+ t.Errorf("Type() = %q, want %q", column.Type(), "OBJECT")
}
- return nil
}
diff --git a/test/e2e/e2e_table_test.go b/test/e2e/e2e_table_test.go
index e56ce95..6355e48 100644
--- a/test/e2e/e2e_table_test.go
+++ b/test/e2e/e2e_table_test.go
@@ -20,6 +20,7 @@
package e2e
import (
+ "bytes"
"log"
"strconv"
"strings"
@@ -424,6 +425,131 @@ func (s *e2eTableTestSuite) Test_InsertTabletAndQuery() {
assert.Equal(int64(8), count)
}
+func (s *e2eTableTestSuite) Test_InsertObjectTablet() {
+ s.T().Skip("OBJECT type is not supported in the e2e environment")
+ assert := s.Require()
+ s.checkError(s.session.ExecuteNonQueryStatement(
+ "create table object_table (region_id string tag, plant_id
string tag, device_id string tag, temperature float field, file object field)"))
+
+ objectBytes := make([]byte, 1024)
+ for i := 0; i < len(objectBytes); i++ {
+ objectBytes[i] = byte(i % 251)
+ }
+ const segmentSize = 512
+ var objectSegments [][]byte
+ for i := 0; i < len(objectBytes); i += segmentSize {
+ end := i + segmentSize
+ if end > len(objectBytes) {
+ end = len(objectBytes)
+ }
+ objectSegments = append(objectSegments, objectBytes[i:end])
+ }
+
+ tablet, err := client.NewRelationalTablet("object_table",
[]*client.MeasurementSchema{
+ {Measurement: "region_id", DataType: client.STRING},
+ {Measurement: "plant_id", DataType: client.STRING},
+ {Measurement: "device_id", DataType: client.STRING},
+ {Measurement: "temperature", DataType: client.FLOAT},
+ {Measurement: "file", DataType: client.OBJECT},
+ }, []client.ColumnCategory{client.TAG, client.TAG, client.TAG,
client.FIELD, client.FIELD}, 1)
+ assert.NoError(err)
+
+ // insert whole object at time 1
+ tablet.SetTimestamp(1, 0)
+ assert.NoError(tablet.SetValueAt("1", 0, 0))
+ assert.NoError(tablet.SetValueAt("5", 1, 0))
+ assert.NoError(tablet.SetValueAt("3", 2, 0))
+ assert.NoError(tablet.SetValueAt(float32(37.6), 3, 0))
+ assert.NoError(tablet.SetObjectValueAt(true, 0, objectBytes, 4, 0))
+ tablet.RowSize++
+ s.checkError(s.session.Insert(tablet))
+ tablet.Reset()
+
+ // insert object in segments at time 2
+ for i, segment := range objectSegments {
+ tablet.SetTimestamp(2, 0)
+ assert.NoError(tablet.SetValueAt("2", 0, 0))
+ assert.NoError(tablet.SetValueAt("6", 1, 0))
+ assert.NoError(tablet.SetValueAt("4", 2, 0))
+ assert.NoError(tablet.SetValueAt(float32(37.7), 3, 0))
+ isEOF := i == len(objectSegments)-1
+ assert.NoError(tablet.SetObjectValueAt(isEOF,
int64(i*segmentSize), segment, 4, 0))
+ tablet.RowSize++
+ s.checkError(s.session.Insert(tablet))
+ tablet.Reset()
+ }
+
+ // insert a row without object value at time 3
+ tablet.SetTimestamp(3, 0)
+ assert.NoError(tablet.SetValueAt("3", 0, 0))
+ assert.NoError(tablet.SetValueAt("7", 1, 0))
+ assert.NoError(tablet.SetValueAt("5", 2, 0))
+ assert.NoError(tablet.SetValueAt(float32(37.8), 3, 0))
+ assert.NoError(tablet.SetValueAt(nil, 4, 0))
+ tablet.RowSize++
+ s.checkError(s.session.Insert(tablet))
+
+ timeoutInMs := int64(10000)
+
+ // count
+ dataSet, err := s.session.ExecuteQueryStatement("select count(*) from
object_table", &timeoutInMs)
+ assert.NoError(err)
+ hasNext, err := dataSet.Next()
+ assert.NoError(err)
+ assert.True(hasNext)
+ count, err := dataSet.GetLongByIndex(1)
+ assert.NoError(err)
+ assert.Equal(int64(3), count)
+ dataSet.Close()
+
+ // read back whole object
+ dataSet, err = s.session.ExecuteQueryStatement("select
READ_OBJECT(file) from object_table where time = 1", &timeoutInMs)
+ assert.NoError(err)
+ hasNext, err = dataSet.Next()
+ assert.NoError(err)
+ assert.True(hasNext)
+ blob, err := dataSet.GetBlobByIndex(1)
+ assert.NoError(err)
+ assert.True(bytes.Equal(objectBytes, blob.GetValues()))
+ dataSet.Close()
+
+ // read back segmented object
+ dataSet, err = s.session.ExecuteQueryStatement("select
READ_OBJECT(file) from object_table where time = 2", &timeoutInMs)
+ assert.NoError(err)
+ hasNext, err = dataSet.Next()
+ assert.NoError(err)
+ assert.True(hasNext)
+ blob, err = dataSet.GetBlobByIndex(1)
+ assert.NoError(err)
+ assert.True(bytes.Equal(objectBytes, blob.GetValues()))
+ dataSet.Close()
+
+ // null object row
+ dataSet, err = s.session.ExecuteQueryStatement("select file from
object_table where time = 3", &timeoutInMs)
+ assert.NoError(err)
+ hasNext, err = dataSet.Next()
+ assert.NoError(err)
+ assert.True(hasNext)
+ isNull, err := dataSet.IsNullByIndex(1)
+ assert.NoError(err)
+ assert.True(isNull)
+ dataSet.Close()
+
+ // non-object columns round-trip
+ dataSet, err = s.session.ExecuteQueryStatement("select region_id,
plant_id, device_id, temperature from object_table where time = 1",
&timeoutInMs)
+ assert.NoError(err)
+ hasNext, err = dataSet.Next()
+ assert.NoError(err)
+ assert.True(hasNext)
+ assert.Equal("1", getValueFromDataSet(dataSet, "region_id"))
+ assert.Equal("5", getValueFromDataSet(dataSet, "plant_id"))
+ assert.Equal("3", getValueFromDataSet(dataSet, "device_id"))
+ temp, err := dataSet.GetFloat("temperature")
+ assert.NoError(err)
+ assert.Equal(float32(37.6), temp)
+ dataSet.Close()
+}
+
func getValueFromDataSet(dataSet *client.SessionDataSet, columnName string)
interface{} {
if isNull, err := dataSet.IsNull(columnName); err != nil {
log.Fatal(err)
