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 "" } @@ -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/object.go b/database/column/object.go new file mode 100644 index 0000000..ebe8232 --- /dev/null +++ b/database/column/object.go @@ -0,0 +1,56 @@ +/* + * 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 column + +import "github.com/apache/iotdb-client-go/v2/client" + +type Object struct { + name string +} + +func (o *Object) Name() string { + return o.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 + } + return "" + } + value, err := stat.GetString(o.name) + if err != nil { + if ptr { + return nil + } + return "" + } + if ptr { + return &value + } + return value +} + +var _ Interface = (*Object)(nil) diff --git a/database/column/object_test.go b/database/column/object_test.go new file mode 100644 index 0000000..8069af4 --- /dev/null +++ b/database/column/object_test.go @@ -0,0 +1,38 @@ +/* + * 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 column + +import "testing" + +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") + } +} 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)