Support OBJECT data type in tablet write

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)
 	}
 

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) {

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 {

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)
+	}
+}

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)
 }
 

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)

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")
+	}
+}