This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch support-object-tablet in repository https://gitbox.apache.org/repos/asf/iotdb-client-go.git
commit 32131c5be90ab65d62dedd6fb10cab4ed816722f Author: shuwenwei <[email protected]> AuthorDate: Thu Aug 20 18:06:10 2026 +0800 Support OBJECT data type in tablet write Add support for the OBJECT data type (TSDataType 12) in the table model tablet write path, following the Java TableSession/tsfile implementation: - Add OBJECT to the TSDataType enum and the string/byte type maps - Treat OBJECT as a binary column (like BLOB) in Tablet: value storage, SetValueAt/GetValueAt, Swap, getValuesBytes and NewTablet - Add Tablet.SetObjectValueAt for segmented OBJECT writes, wrapping each segment with a 1-byte isEOF flag and an 8-byte big-endian offset, matching Java Tablet.addValue(rowIndex, columnIndex, isEOF, offset, content) - Decode OBJECT (and BLOB/STRING) binary columns in the read path so written objects can be read back (e.g. via READ_OBJECT) - Add unit tests for whole-object and segmented OBJECT tablet writes - Add an e2e table test writing OBJECT via tablet, covering whole-object, segmented and null-object rows Also include the regenerated thrift common code (new aggregation types, pipeRecentFailureList field) from `make all`. --- client/column_decoder.go | 2 +- client/protocol.go | 3 + client/rpcdataset.go | 4 +- client/tablet.go | 46 ++++++++-- client/tablet_test.go | 93 ++++++++++++++++++++ common/common.go | 205 ++++++++++++++++++++++++++++++++++++--------- test/e2e/e2e_table_test.go | 125 +++++++++++++++++++++++++++ 7 files changed, 429 insertions(+), 49 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..6d3df02 100644 --- a/client/rpcdataset.go +++ b/client/rpcdataset.go @@ -542,7 +542,7 @@ func (s *IoTDBRpcDataSet) getObjectByTsBlockIndex(tsBlockColumnIndex int32) (int } else { return binary.GetStringValue(), nil } - case BLOB: + case BLOB, OBJECT: if binary, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBinary(s.tsBlockIndex); err != nil { return nil, err } else { @@ -639,7 +639,7 @@ func (s *IoTDBRpcDataSet) getStringByTsBlockColumnIndexAndDataType(index int32, } else { return v.GetStringValue(), nil } - case BLOB: + case BLOB, OBJECT: if v, err := s.curTsBlock.GetColumn(index).GetBinary(s.tsBlockIndex); err != nil { return "", err } else { diff --git a/client/tablet.go b/client/tablet.go index d694d6f..90b6b11 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] } @@ -195,7 +195,7 @@ func (t *Tablet) SetValueAt(value interface{}, columnIndex, rowIndex int) error default: return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value)) } - case TEXT, STRING: + case TEXT, STRING, OBJECT: values := t.values[columnIndex].([][]byte) switch v := value.(type) { case string: @@ -229,6 +229,42 @@ func (t *Tablet) SetValueAt(value interface{}, columnIndex, rowIndex int) error return nil } +// 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) + return t.SetValueAt(value, columnIndex, rowIndex) +} + func (t *Tablet) GetMaxRowNumber() int { return t.maxRowNumber } @@ -260,7 +296,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 +349,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 +401,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..d15803c 100644 --- a/client/tablet_test.go +++ b/client/tablet_test.go @@ -676,3 +676,96 @@ 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.Fatalf("SetValueAt([]byte) error = %v", err) + } + tablet.SetTimestamp(1608268702780, 0) + tablet.RowSize++ + + if err := tablet.SetValueAt("hello", 1, 1); err != nil { + t.Fatalf("SetValueAt(string) error = %v", err) + } + tablet.SetTimestamp(1608268702781, 1) + tablet.RowSize++ + + if got, err := tablet.GetValueAt(1, 0); err != nil || !reflect.DeepEqual(got, objVal) { + t.Errorf("GetValueAt(1,0) = %v, %v; want %v, nil", got, err, objVal) + } + if got, err := tablet.GetValueAt(1, 1); err != nil || !reflect.DeepEqual(got, []byte("hello")) { + t.Errorf("GetValueAt(1,1) = %v, %v; want %v, nil", got, err, []byte("hello")) + } + + valuesBytes, err := tablet.getValuesBytes() + if err != nil { + t.Fatal(err) + } + wantValues := []byte{ + 0, 0, 0, 0, + 0, 0, 0, 0, + 0, 0, 0, 3, 0x01, 0x02, 0x03, + 0, 0, 0, 5, 'h', 'e', 'l', 'l', 'o', + } + 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.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/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/test/e2e/e2e_table_test.go b/test/e2e/e2e_table_test.go index e56ce95..63b4ff2 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,130 @@ func (s *e2eTableTestSuite) Test_InsertTabletAndQuery() { assert.Equal(int64(8), count) } +func (s *e2eTableTestSuite) Test_InsertObjectTablet() { + 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)
