github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-21 08:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(iotdb): align 1.3.x aggregate columns and format max_time

Browse Source 
Closes #7306
This commit is contained in:
t8y2
2026-09-01 15:18:44 +08:00
parent 6588f27fbc
commit 613c4f7ca3
34 changed files with 64422 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/drivers/iotdb/README.md
+7
View File
@@ -51,6 +51,13 @@ for TLS or mTLS connections.
- Query results annotate timestamp columns as `TIMESTAMP(ms|us|ns)` using the
server-reported `TimestampPrecision`. Raw timestamp integers are transported
as decimal strings so nanosecond values are not rounded by JavaScript.
- Tree `max_time`/`min_time` aggregates also return epoch values; they are
annotated as `TIMESTAMP(ms|us|ns)` too so the grid renders standard times.
- The pinned client is vendored under `agents/go-common/iotdb-client-go` with
one patch: upstream drops the 1.3.x `ColumnNameIndexMap` fallback, which
misaligned aggregate result values against the SELECT column order on 1.3.x
servers (the patched client maps value columns through the server-provided
name index when the ordered index list is absent).
- The Agent keeps one physical IoTDB session per logical DBX session and
invalidates that session after cancellation, timeout, or connection failure.
agents/drivers/iotdb/go.mod
+4
View File
@@ -5,3 +5,7 @@ go 1.25
require github.com/apache/iotdb-client-go/v2 v2.0.9-0.20260807074554-e59fc7f55df1
require github.com/apache/thrift v0.24.0 // indirect
// Upstream ignores the 1.3.x ColumnNameIndexMap fallback, which misaligns
// aggregate result columns against the SELECT list on 1.3.x servers.
replace github.com/apache/iotdb-client-go/v2 => ../../go-common/iotdb-client-go
agents/drivers/iotdb/go.sum
-2
View File
@@ -1,4 +1,2 @@
github.com/apache/iotdb-client-go/v2 v2.0.9-0.20260807074554-e59fc7f55df1 h1:UAfvet8VA2ZUCsIBusSqNDDecFGot8uDshegeDdLBRo=
github.com/apache/iotdb-client-go/v2 v2.0.9-0.20260807074554-e59fc7f55df1/go.mod h1:Ef61z8w2tRnuAcQKESFfiH03kDvfQniVPUgOXCxWnNw=
github.com/apache/thrift v0.24.0 h1:zy31L1a49QTNB2bG1BBfMXol3yJrTH975G3pPubQVLQ=
github.com/apache/thrift v0.24.0/go.mod h1:zPt6WxgvTOM6hF92y8C+MkEM5LMxZuk4JcQOiU4Esvs=
agents/drivers/iotdb/integration_test.go
+19
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"os"
"os/exec"
"reflect"
"strconv"
"strings"
"sync"
@@ -31,6 +32,24 @@ func TestLiveIoTDBAgentTreeAndTable(t *testing.T) {
mustExecuteNonQuery(t, treeServer, fmt.Sprintf("INSERT INTO %s(time,s1) VALUES(%d,%d)", treeDevice, index, index*10), "")
}
// 1.3.x servers order aggregate TsBlock columns by their own aggregate
// layout, not the SELECT list; values must still line up with the headers
// (https://github.com/t8y2/dbx/issues/7306).
aggregate, err := treeServer.executeQuery(queryOptions{
SQL: "SELECT max_time(s1), avg(s1), max_value(s1), min_value(s1) FROM " + treeDevice,
MaxRows: 10,
})
if err != nil || len(aggregate.Rows) != 1 {
t.Fatalf("tree aggregate = %#v, %v", aggregate, err)
}
wantAggregateRow := []any{"4", 25.0, int64(40), int64(10)}
if !reflect.DeepEqual(aggregate.Rows[0], wantAggregateRow) {
t.Fatalf("tree aggregate row = %#v, want %#v", aggregate.Rows[0], wantAggregateRow)
}
if aggregate.ColumnTypes[0] != "TIMESTAMP(ms)" {
t.Fatalf("tree aggregate max_time type = %#v", aggregate.ColumnTypes)
}
tables, err := treeServer.listTables(treeDatabase, metadataListConstraints{})
if err != nil || len(tables) != 1 || tables[0].Name != "d1" {
t.Fatalf("tree listTables() = %#v, %v", tables, err)
agents/drivers/iotdb/main_test.go
+21
View File
@@ -178,6 +178,27 @@ func TestTreeTimeColumnPresentationDoesNotRewriteOrdinaryInt64(t *testing.T) {
}
}
func TestTreeTimeAggregationColumnsPresentAsTimestamps(t *testing.T) {
columns := []string{
"max_time(root.db.d1.s1)",
"MIN_TIME(root.db.d1.s1)",
" max_time(root.db.d1.s1) ",
"avg(root.db.d1.s1)",
"count(root.db.d1.s1)",
}
got := normalizedColumnTypes([]string{"INT64", "INT64", "INT64", "DOUBLE", "INT64"}, columns, client.TreeSqlDialect, "ms")
want := []string{"TIMESTAMP(ms)", "TIMESTAMP(ms)", "TIMESTAMP(ms)", "DOUBLE", "INT64"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("unexpected tree aggregation types: %#v", got)
}
if name := normalizedColumnTypes([]string{"INT64"}, []string{"max_timer(root.db.d1.s1)"}, client.TreeSqlDialect, "ms"); !reflect.DeepEqual(name, []string{"INT64"}) {
t.Fatalf("unrelated prefix must stay INT64: %#v", name)
}
if table := normalizedColumnTypes([]string{"INT64"}, []string{"max_time(t1.s1)"}, client.TableSqlDialect, "ms"); !reflect.DeepEqual(table, []string{"INT64"}) {
t.Fatalf("table dialect must stay INT64: %#v", table)
}
}
func TestClassifyIoTDBErrors(t *testing.T) {
canceled := classifyRPCError("execute_query", "session-1", context.Canceled)
if canceled.Data.Category != "canceled" || canceled.Data.SessionDisposition != "quarantine" {
agents/drivers/iotdb/query.go
+16
View File
@@ -475,11 +475,27 @@ func normalizedColumnTypes(values, columns []string, dialect, timestampPrecision
}
if iotdbColumnTypeBase(result[index]) == "TIMESTAMP" {
result[index] = timestampColumnType(timestampPrecision)
continue
}
// max_time/min_time aggregates return INT64 epoch values in the tree
// dialect; type them as timestamps so the grid renders standard times.
if dialect == client.TreeSqlDialect && iotdbColumnTypeBase(result[index]) == "INT64" && isTimeAggregationColumn(columns[index]) {
result[index] = timestampColumnType(timestampPrecision)
}
}
return result
}
func isTimeAggregationColumn(column string) bool {
trimmed := strings.TrimSpace(column)
for _, prefix := range []string{"max_time(", "min_time("} {
if len(trimmed) > len(prefix) && strings.EqualFold(trimmed[:len(prefix)], prefix) {
return true
}
}
return false
}
func timestampColumnType(precision string) string {
if normalized := normalizeTimestampPrecision(precision); normalized != "" {
return "TIMESTAMP(" + normalized + ")"
agents/go-common/iotdb-client-go/LICENSE
+207
View File
@@ -0,0 +1,207 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed 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.
--------------------------------------------------------------------------------
APACHE IOTDB SUBCOMPONENTS
--------------------------------------------------------------------------------
agents/go-common/iotdb-client-go/NOTICE
+19
View File
@@ -0,0 +1,19 @@
Apache IoTDB
Copyright 2018-2020 The Apache Software Foundation.
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
============================================================================
IoTDB project uses 4 Chinese Patents:
* 201711384490X
* 201810111712.9
* 201711322631.5
* 201711319331.1
According to the Apache 2.0 License. The owner of the patents, Tsinghua University,
grant the users the right to the use of patent under the requirement of Apache 2.0 License.
============================================================================
agents/go-common/iotdb-client-go/client/bitmap.go
+76
View File
@@ -0,0 +1,76 @@
/*
* 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
type BitMap struct {
size int
bits []byte
}
var BitUtil = []byte{1, 2, 4, 8, 16, 32, 64, 128}
var UnmarkBitUtil = []byte{
0xFE, // 11111110
0xFD, // 11111101
0xFB, // 11111011
0xF7, // 11110111
0xEF, // 11101111
0xDF, // 11011111
0xBF, // 10111111
0x7F, // 01111111
}
func NewBitMap(size int) *BitMap {
// Need to maintain consistency with the calculation method on the IoTDB side.
bitMap := &BitMap{
size: size,
bits: make([]byte, size/8+1),
}
return bitMap
}
func (b *BitMap) Mark(position int) {
b.bits[position/8] |= BitUtil[position%8]
}
func (b *BitMap) UnMark(position int) {
b.bits[position/8] &= UnmarkBitUtil[position%8]
}
func (b *BitMap) IsMarked(position int) bool {
return (b.bits[position/8] & BitUtil[position%8]) != 0
}
func (b *BitMap) IsAllUnmarked() bool {
for i := 0; i < b.size/8; i++ {
if b.bits[i] != 0 {
return false
}
}
for i := 0; i < b.size%8; i++ {
if (b.bits[b.size/8] & BitUtil[i]) != 0 {
return false
}
}
return true
}
func (b *BitMap) GetBits() []byte {
return b.bits
}
agents/go-common/iotdb-client-go/client/column.go
+821
View File
@@ -0,0 +1,821 @@
/*
* 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 "fmt"
type ColumnEncoding uint8
const (
BYTE_ARRAY_COLUMN_ENCODING = ColumnEncoding(iota)
INT32_ARRAY_COLUMN_ENCODING
INT64_ARRAY_COLUMN_ENCODING
BINARY_ARRAY_COLUMN_ENCODING
RLE_COLUMN_ENCODING
)
var encodingToDecoder = map[ColumnEncoding]ColumnDecoder{
INT32_ARRAY_COLUMN_ENCODING: new(Int32ArrayColumnDecoder),
INT64_ARRAY_COLUMN_ENCODING: new(Int64ArrayColumnDecoder),
BYTE_ARRAY_COLUMN_ENCODING: new(ByteArrayColumnDecoder),
BINARY_ARRAY_COLUMN_ENCODING: new(BinaryArrayColumnDecoder),
RLE_COLUMN_ENCODING: new(RunLengthColumnDecoder),
}
var byteToEncoding = map[byte]ColumnEncoding{
0: BYTE_ARRAY_COLUMN_ENCODING,
1: INT32_ARRAY_COLUMN_ENCODING,
2: INT64_ARRAY_COLUMN_ENCODING,
3: BINARY_ARRAY_COLUMN_ENCODING,
4: RLE_COLUMN_ENCODING,
}
func getColumnDecoder(encoding ColumnEncoding) (ColumnDecoder, error) {
decoder, exists := encodingToDecoder[encoding]
if !exists {
return nil, fmt.Errorf("unsupported column encoding: %v", encoding)
}
return decoder, nil
}
func getColumnEncodingByByte(b byte) (ColumnEncoding, error) {
encoding, exists := byteToEncoding[b]
if !exists {
return INT32_ARRAY_COLUMN_ENCODING, fmt.Errorf("invalid value: %v", b)
}
return encoding, nil
}
type Column interface {
GetDataType() TSDataType
GetEncoding() ColumnEncoding
GetBoolean(position int32) (bool, error)
GetInt(position int32) (int32, error)
GetLong(position int32) (int64, error)
GetFloat(position int32) (float32, error)
GetDouble(position int32) (float64, error)
GetBinary(position int32) (*Binary, error)
GetObject(position int32) (interface{}, error)
GetBooleans() ([]bool, error)
GetInts() ([]int32, error)
GetLongs() ([]int64, error)
GetFloats() ([]float32, error)
GetDoubles() ([]float64, error)
GetBinaries() ([]*Binary, error)
GetObjects() ([]interface{}, error)
MayHaveNull() bool
IsNull(position int32) bool
IsNulls() []bool
GetPositionCount() int32
}
type baseColumn struct {
}
func (c *baseColumn) GetBoolean(_ int32) (bool, error) {
return false, fmt.Errorf("unsupported operation: GetBoolean")
}
func (c *baseColumn) GetInt(_ int32) (int32, error) {
return 0, fmt.Errorf("unsupported operation: GetInt")
}
func (c *baseColumn) GetLong(_ int32) (int64, error) {
return 0, fmt.Errorf("unsupported operation: GetLong")
}
func (c *baseColumn) GetFloat(_ int32) (float32, error) {
return 0, fmt.Errorf("unsupported operation: GetFloat")
}
func (c *baseColumn) GetDouble(_ int32) (float64, error) {
return 0, fmt.Errorf("unsupported operation: GetDouble")
}
func (c *baseColumn) GetBinary(_ int32) (*Binary, error) {
return nil, fmt.Errorf("unsupported operation: GetBinary")
}
func (c *baseColumn) GetObject(_ int32) (interface{}, error) {
return nil, fmt.Errorf("unsupported operation: GetObject")
}
func (c *baseColumn) GetBooleans() ([]bool, error) {
return nil, fmt.Errorf("unsupported operation: GetBooleans")
}
func (c *baseColumn) GetInts() ([]int32, error) {
return nil, fmt.Errorf("unsupported operation: GetInts")
}
func (c *baseColumn) GetLongs() ([]int64, error) {
return nil, fmt.Errorf("unsupported operation: GetLongs")
}
func (c *baseColumn) GetFloats() ([]float32, error) {
return nil, fmt.Errorf("unsupported operation: GetFloats")
}
func (c *baseColumn) GetDoubles() ([]float64, error) {
return nil, fmt.Errorf("unsupported operation: GetDoubles")
}
func (c *baseColumn) GetBinaries() ([]*Binary, error) {
return nil, fmt.Errorf("unsupported operation: GetBinaries")
}
func (c *baseColumn) GetObjects() ([]interface{}, error) {
return nil, fmt.Errorf("unsupported operation: GetObjects")
}
type TimeColumn struct {
baseColumn
arrayOffset int32
positionCount int32
values []int64
}
func NewTimeColumn(arrayOffset int32, positionCount int32, values []int64) (*TimeColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
return &TimeColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
values: values,
}, nil
}
func (tc *TimeColumn) GetDataType() TSDataType {
return INT64
}
func (tc *TimeColumn) GetEncoding() ColumnEncoding {
return INT64_ARRAY_COLUMN_ENCODING
}
func (tc *TimeColumn) GetLong(position int32) (int64, error) {
return tc.values[position+tc.arrayOffset], nil
}
func (tc *TimeColumn) MayHaveNull() bool {
return false
}
func (tc *TimeColumn) IsNull(_ int32) bool {
return false
}
func (tc *TimeColumn) IsNulls() []bool {
return nil
}
func (tc *TimeColumn) GetPositionCount() int32 {
return tc.positionCount
}
func (tc *TimeColumn) GetStartTime() int64 {
return tc.values[tc.arrayOffset]
}
func (tc *TimeColumn) GetEndTime() int64 {
return tc.values[tc.positionCount+tc.arrayOffset-1]
}
func (tc *TimeColumn) GetTimes() []int64 {
return tc.values
}
func (tc *TimeColumn) GetLongs() ([]int64, error) {
return tc.GetTimes(), nil
}
type BinaryColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []*Binary
}
func NewBinaryColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []*Binary) (*BinaryColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &BinaryColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *BinaryColumn) GetDataType() TSDataType {
return TEXT
}
func (c *BinaryColumn) GetEncoding() ColumnEncoding {
return BINARY_ARRAY_COLUMN_ENCODING
}
func (c *BinaryColumn) GetBinary(position int32) (*Binary, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *BinaryColumn) GetBinaries() ([]*Binary, error) {
return c.values, nil
}
func (c *BinaryColumn) GetObject(position int32) (interface{}, error) {
return c.GetBinary(position)
}
func (c *BinaryColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *BinaryColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *BinaryColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *BinaryColumn) GetPositionCount() int32 {
return c.positionCount
}
type IntColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []int32
}
func NewIntColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []int32) (*IntColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &IntColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *IntColumn) GetDataType() TSDataType {
return INT32
}
func (c *IntColumn) GetEncoding() ColumnEncoding {
return INT32_ARRAY_COLUMN_ENCODING
}
func (c *IntColumn) GetInt(position int32) (int32, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *IntColumn) GetInts() ([]int32, error) {
return c.values, nil
}
func (c *IntColumn) GetObject(position int32) (interface{}, error) {
return c.GetInt(position)
}
func (c *IntColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *IntColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *IntColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *IntColumn) GetPositionCount() int32 {
return c.positionCount
}
type FloatColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []float32
}
func NewFloatColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []float32) (*FloatColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &FloatColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *FloatColumn) GetDataType() TSDataType {
return FLOAT
}
func (c *FloatColumn) GetEncoding() ColumnEncoding {
return INT32_ARRAY_COLUMN_ENCODING
}
func (c *FloatColumn) GetFloat(position int32) (float32, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *FloatColumn) GetFloats() ([]float32, error) {
return c.values, nil
}
func (c *FloatColumn) GetObject(position int32) (interface{}, error) {
return c.GetFloat(position)
}
func (c *FloatColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *FloatColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *FloatColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *FloatColumn) GetPositionCount() int32 {
return c.positionCount
}
type LongColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []int64
}
func NewLongColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []int64) (*LongColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &LongColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *LongColumn) GetDataType() TSDataType {
return INT64
}
func (c *LongColumn) GetEncoding() ColumnEncoding {
return INT64_ARRAY_COLUMN_ENCODING
}
func (c *LongColumn) GetLong(position int32) (int64, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *LongColumn) GetLongs() ([]int64, error) {
return c.values, nil
}
func (c *LongColumn) GetObject(position int32) (interface{}, error) {
return c.GetLong(position)
}
func (c *LongColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *LongColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *LongColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *LongColumn) GetPositionCount() int32 {
return c.positionCount
}
type DoubleColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []float64
}
func NewDoubleColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []float64) (*DoubleColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &DoubleColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *DoubleColumn) GetDataType() TSDataType {
return DOUBLE
}
func (c *DoubleColumn) GetEncoding() ColumnEncoding {
return INT64_ARRAY_COLUMN_ENCODING
}
func (c *DoubleColumn) GetDouble(position int32) (float64, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *DoubleColumn) GetDoubles() ([]float64, error) {
return c.values, nil
}
func (c *DoubleColumn) GetObject(position int32) (interface{}, error) {
return c.GetDouble(position)
}
func (c *DoubleColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *DoubleColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *DoubleColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *DoubleColumn) GetPositionCount() int32 {
return c.positionCount
}
type BooleanColumn struct {
baseColumn
arrayOffset int32
positionCount int32
valueIsNull []bool
values []bool
}
func NewBooleanColumn(arrayOffset int32, positionCount int32, valueIsNull []bool, values []bool) (*BooleanColumn, error) {
if arrayOffset < 0 {
return nil, fmt.Errorf("arrayOffset is negative")
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
if int32(len(values))-arrayOffset < positionCount {
return nil, fmt.Errorf("values length is less than positionCount")
}
if valueIsNull != nil && int32(len(valueIsNull))-arrayOffset < positionCount {
return nil, fmt.Errorf("isNull length is less than positionCount")
}
return &BooleanColumn{
arrayOffset: arrayOffset,
positionCount: positionCount,
valueIsNull: valueIsNull,
values: values,
}, nil
}
func (c *BooleanColumn) GetDataType() TSDataType {
return BOOLEAN
}
func (c *BooleanColumn) GetEncoding() ColumnEncoding {
return BYTE_ARRAY_COLUMN_ENCODING
}
func (c *BooleanColumn) GetBoolean(position int32) (bool, error) {
return c.values[position+c.arrayOffset], nil
}
func (c *BooleanColumn) GetBooleans() ([]bool, error) {
return c.values, nil
}
func (c *BooleanColumn) GetObject(position int32) (interface{}, error) {
return c.GetBoolean(position)
}
func (c *BooleanColumn) MayHaveNull() bool {
return c.valueIsNull != nil
}
func (c *BooleanColumn) IsNull(position int32) bool {
return c.valueIsNull != nil && c.valueIsNull[position+c.arrayOffset]
}
func (c *BooleanColumn) IsNulls() []bool {
if c.valueIsNull != nil {
return c.valueIsNull
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = false
}
return result
}
func (c *BooleanColumn) GetPositionCount() int32 {
return c.positionCount
}
type RunLengthEncodedColumn struct {
baseColumn
value Column
positionCount int32
}
func NewRunLengthEncodedColumn(value Column, positionCount int32) (*RunLengthEncodedColumn, error) {
if value == nil {
return nil, fmt.Errorf("value is null")
}
if value.GetPositionCount() != 1 {
return nil, fmt.Errorf("expected value to contain a single position but has %v positions", value.GetPositionCount())
}
if positionCount < 0 {
return nil, fmt.Errorf("positionCount is negative")
}
column := new(RunLengthEncodedColumn)
switch (value).(type) {
case *RunLengthEncodedColumn:
column.value = (value.(*RunLengthEncodedColumn)).GetValue()
default:
column.value = value
}
column.positionCount = positionCount
return column, nil
}
func (c *RunLengthEncodedColumn) GetValue() Column {
return c.value
}
func (c *RunLengthEncodedColumn) GetDataType() TSDataType {
return c.value.GetDataType()
}
func (c *RunLengthEncodedColumn) GetEncoding() ColumnEncoding {
return RLE_COLUMN_ENCODING
}
func (c *RunLengthEncodedColumn) GetBoolean(_ int32) (bool, error) {
return c.value.GetBoolean(0)
}
func (c *RunLengthEncodedColumn) GetInt(_ int32) (int32, error) {
return c.value.GetInt(0)
}
func (c *RunLengthEncodedColumn) GetLong(_ int32) (int64, error) {
return c.value.GetLong(0)
}
func (c *RunLengthEncodedColumn) GetFloat(_ int32) (float32, error) {
return c.value.GetFloat(0)
}
func (c *RunLengthEncodedColumn) GetDouble(_ int32) (float64, error) {
return c.value.GetDouble(0)
}
func (c *RunLengthEncodedColumn) GetBinary(_ int32) (*Binary, error) {
return c.value.GetBinary(0)
}
func (c *RunLengthEncodedColumn) GetObject(_ int32) (interface{}, error) {
return c.value.GetObject(0)
}
func (c *RunLengthEncodedColumn) GetBooleans() ([]bool, error) {
v, err := c.value.GetBoolean(0)
if err != nil {
return nil, err
}
result := make([]bool, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetInts() ([]int32, error) {
v, err := c.value.GetInt(0)
if err != nil {
return nil, err
}
result := make([]int32, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetLongs() ([]int64, error) {
v, err := c.value.GetLong(0)
if err != nil {
return nil, err
}
result := make([]int64, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetFloats() ([]float32, error) {
v, err := c.value.GetFloat(0)
if err != nil {
return nil, err
}
result := make([]float32, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetDoubles() ([]float64, error) {
v, err := c.value.GetDouble(0)
if err != nil {
return nil, err
}
result := make([]float64, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetBinaries() ([]*Binary, error) {
v, err := c.value.GetBinary(0)
if err != nil {
return nil, err
}
result := make([]*Binary, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) GetObjects() ([]interface{}, error) {
v, err := c.value.GetObject(0)
if err != nil {
return nil, err
}
result := make([]interface{}, c.positionCount)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result, err
}
func (c *RunLengthEncodedColumn) MayHaveNull() bool {
return c.value.MayHaveNull()
}
func (c *RunLengthEncodedColumn) IsNull(_ int32) bool {
return c.value.IsNull(0)
}
func (c *RunLengthEncodedColumn) IsNulls() []bool {
result := make([]bool, c.positionCount)
v := c.value.IsNull(0)
for i := int32(0); i < c.positionCount; i++ {
result[i] = v
}
return result
}
func (c *RunLengthEncodedColumn) GetPositionCount() int32 {
return c.positionCount
}
agents/go-common/iotdb-client-go/client/column_decoder.go
+308
View File
@@ -0,0 +1,308 @@
/*
* 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 (
"bytes"
"encoding/binary"
"fmt"
)
type ColumnDecoder interface {
ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error)
}
func deserializeNullIndicators(reader *bytes.Reader, positionCount int32) ([]bool, error) {
b, err := reader.ReadByte()
if err != nil {
return nil, err
}
mayHaveNull := b != 0
if !mayHaveNull {
return nil, nil
}
return deserializeBooleanArray(reader, positionCount)
}
func deserializeBooleanArray(reader *bytes.Reader, size int32) ([]bool, error) {
packedSize := (size + 7) / 8
packedBytes := make([]byte, packedSize)
_, err := reader.Read(packedBytes)
if err != nil {
return nil, err
}
// read null bits 8 at a time
output := make([]bool, size)
currentByte := 0
fullGroups := int(size) & ^0b111
for pos := 0; pos < fullGroups; pos += 8 {
b := packedBytes[currentByte]
currentByte++
output[pos+0] = (b & 0b10000000) != 0
output[pos+1] = (b & 0b01000000) != 0
output[pos+2] = (b & 0b00100000) != 0
output[pos+3] = (b & 0b00010000) != 0
output[pos+4] = (b & 0b00001000) != 0
output[pos+5] = (b & 0b00000100) != 0
output[pos+6] = (b & 0b00000010) != 0
output[pos+7] = (b & 0b00000001) != 0
}
// read last null bits
if remaining := int(size) % 8; remaining > 0 {
b := packedBytes[len(packedBytes)-1]
mask := uint8(0b10000000)
for pos := fullGroups; pos < int(size); pos++ {
output[pos] = (b & mask) != 0
mask >>= 1
}
}
return output, nil
}
type baseColumnDecoder struct{}
type Int32ArrayColumnDecoder struct {
baseColumnDecoder
}
func (decoder *Int32ArrayColumnDecoder) ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error) {
// Serialized data layout:
// +---------------+-----------------+-------------+
// | may have null | null indicators | values |
// +---------------+-----------------+-------------+
// | byte | list[byte] | list[int32] |
// +---------------+-----------------+-------------+
if positionCount == 0 {
switch dataType {
case INT32, DATE:
return NewIntColumn(0, 0, nil, []int32{})
case FLOAT:
return NewFloatColumn(0, 0, nil, []float32{})
default:
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
}
nullIndicators, err := deserializeNullIndicators(reader, positionCount)
if err != nil {
return nil, err
}
switch dataType {
case INT32, DATE:
intValues := make([]int32, positionCount)
for i := int32(0); i < positionCount; i++ {
if nullIndicators != nil && nullIndicators[i] {
continue
}
err := binary.Read(reader, binary.BigEndian, &intValues[i])
if err != nil {
return nil, err
}
}
return NewIntColumn(0, positionCount, nullIndicators, intValues)
case FLOAT:
floatValues := make([]float32, positionCount)
for i := int32(0); i < positionCount; i++ {
if nullIndicators != nil && nullIndicators[i] {
continue
}
err := binary.Read(reader, binary.BigEndian, &floatValues[i])
if err != nil {
return nil, err
}
}
return NewFloatColumn(0, positionCount, nullIndicators, floatValues)
}
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
type Int64ArrayColumnDecoder struct {
baseColumnDecoder
}
func (decoder *Int64ArrayColumnDecoder) ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error) {
// Serialized data layout:
// +---------------+-----------------+-------------+
// | may have null | null indicators | values |
// +---------------+-----------------+-------------+
// | byte | list[byte] | list[int64] |
// +---------------+-----------------+-------------+
if positionCount == 0 {
switch dataType {
case INT64, TIMESTAMP:
return NewLongColumn(0, 0, nil, []int64{})
case DOUBLE:
return NewDoubleColumn(0, 0, nil, []float64{})
default:
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
}
nullIndicators, err := deserializeNullIndicators(reader, positionCount)
if err != nil {
return nil, err
}
switch dataType {
case INT64, TIMESTAMP:
values := make([]int64, positionCount)
for i := int32(0); i < positionCount; i++ {
if nullIndicators != nil && nullIndicators[i] {
continue
}
if err = binary.Read(reader, binary.BigEndian, &values[i]); err != nil {
return nil, err
}
}
return NewLongColumn(0, positionCount, nullIndicators, values)
case DOUBLE:
values := make([]float64, positionCount)
for i := int32(0); i < positionCount; i++ {
if nullIndicators != nil && nullIndicators[i] {
continue
}
if err = binary.Read(reader, binary.BigEndian, &values[i]); err != nil {
return nil, err
}
}
return NewDoubleColumn(0, positionCount, nullIndicators, values)
}
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
type ByteArrayColumnDecoder struct {
baseColumnDecoder
}
func (decoder *ByteArrayColumnDecoder) ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error) {
// Serialized data layout:
// +---------------+-----------------+-------------+
// | may have null | null indicators | values |
// +---------------+-----------------+-------------+
// | byte | list[byte] | list[byte] |
// +---------------+-----------------+-------------+
if dataType != BOOLEAN {
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
if positionCount == 0 {
return NewBooleanColumn(0, 0, nil, []bool{})
}
nullIndicators, err := deserializeNullIndicators(reader, positionCount)
if err != nil {
return nil, err
}
values, err := deserializeBooleanArray(reader, positionCount)
if err != nil {
return nil, err
}
return NewBooleanColumn(0, positionCount, nullIndicators, values)
}
type BinaryArrayColumnDecoder struct {
baseColumnDecoder
}
func (decoder *BinaryArrayColumnDecoder) ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error) {
// Serialized data layout:
// +---------------+-----------------+-------------+
// | may have null | null indicators | values |
// +---------------+-----------------+-------------+
// | byte | list[byte] | list[entry] |
// +---------------+-----------------+-------------+
//
// Each entry is represented as:
// +---------------+-------+
// | value length | value |
// +---------------+-------+
// | int32 | bytes |
// +---------------+-------+
if TEXT != dataType {
return nil, fmt.Errorf("invalid data type: %v", dataType)
}
if positionCount == 0 {
return NewBinaryColumn(0, 0, nil, []*Binary{})
}
nullIndicators, err := deserializeNullIndicators(reader, positionCount)
if err != nil {
return nil, err
}
values := make([]*Binary, positionCount)
for i := int32(0); i < positionCount; i++ {
if nullIndicators != nil && nullIndicators[i] {
continue
}
var length int32
err := binary.Read(reader, binary.BigEndian, &length)
if err != nil {
return nil, err
}
if length == 0 {
values[i] = NewBinary([]byte{})
} else {
value := make([]byte, length)
_, err = reader.Read(value)
if err != nil {
return nil, err
}
values[i] = NewBinary(value)
}
}
return NewBinaryColumn(0, positionCount, nullIndicators, values)
}
type RunLengthColumnDecoder struct {
baseColumnDecoder
}
func (decoder *RunLengthColumnDecoder) ReadColumn(reader *bytes.Reader, dataType TSDataType, positionCount int32) (Column, error) {
// Serialized data layout:
// +-----------+-------------------------+
// | encoding | serialized inner column |
// +-----------+-------------------------+
// | byte | list[byte] |
// +-----------+-------------------------+
columnEncoding, err := deserializeColumnEncoding(reader)
if err != nil {
return nil, err
}
columnDecoder, err := getColumnDecoder(columnEncoding)
if err != nil {
return nil, err
}
column, err := columnDecoder.ReadColumn(reader, dataType, 1)
if err != nil {
return nil, err
}
return NewRunLengthEncodedColumn(column, positionCount)
}
agents/go-common/iotdb-client-go/client/errors.go
+53
View File
@@ -0,0 +1,53 @@
/*
* 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 (
"fmt"
"github.com/apache/iotdb-client-go/v2/common"
)
// ExecutionError represents an error returned by the server via TSStatus.
// It is NOT a connection error and should not cause session drops.
type ExecutionError struct {
Code int32
Message string
}
func (e *ExecutionError) Error() string {
if e.Message != "" {
return fmt.Sprintf("error code: %d, message: %v", e.Code, e.Message)
}
return fmt.Sprintf("error code: %d", e.Code)
}
type BatchError struct {
statuses []*common.TSStatus
Message string
}
func (e *BatchError) Error() string {
return e.Message
}
func (e *BatchError) GetStatuses() []*common.TSStatus {
return e.statuses
}
agents/go-common/iotdb-client-go/client/field.go
+100
View File
@@ -0,0 +1,100 @@
/*
* 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 "time"
type Field struct {
dataType TSDataType
name string
value interface{}
}
func (f *Field) IsNull() bool {
return f.value == nil
}
func (f *Field) GetDataType() TSDataType {
return f.dataType
}
func (f *Field) GetValue() interface{} {
return f.value
}
func (f *Field) GetInt32() int32 {
if f.value == nil {
return 0
}
return f.value.(int32)
}
func (f *Field) GetInt64() int64 {
if f.value == nil {
return 0
}
return f.value.(int64)
}
func (f *Field) GetFloat32() float32 {
if f.value == nil {
return 0
}
return f.value.(float32)
}
func (f *Field) GetFloat64() float64 {
if f.value == nil {
return 0
}
return f.value.(float64)
}
func (f *Field) GetName() string {
return f.name
}
func (f *Field) GetText() string {
if f.value == nil {
return ""
}
switch f.value.(type) {
case bool:
if f.value.(bool) {
return "true"
}
return "false"
case int32:
return int32ToString(f.value.(int32))
case int64:
return int64ToString(f.value.(int64))
case float32:
return float32ToString(f.value.(float32))
case float64:
return float64ToString(f.value.(float64))
case string:
return f.value.(string)
case []byte:
return bytesToHexString(f.value.([]byte))
case time.Time:
return f.value.(time.Time).Format("2006-01-02")
}
return ""
}
agents/go-common/iotdb-client-go/client/options.go
+49
View File
@@ -0,0 +1,49 @@
/*
* 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 "context"
// Options holds the configuration for a client call.
type Options struct {
ctx context.Context
}
// Option is a functional option for configuring [Options].
type Option func(*Options)
// ApplyOptions applies the given options and returns the resulting Options.
func ApplyOptions(opts ...Option) Options {
var result Options
for _, opt := range opts {
if opt == nil {
continue
}
opt(&result)
}
return result
}
// WithCtx returns an Option that sets the context for a client call.
func WithCtx(ctx context.Context) Option {
return func(opts *Options) {
opts.ctx = ctx
}
}
agents/go-common/iotdb-client-go/client/protocol.go
+252
View File
@@ -0,0 +1,252 @@
/*
* 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 "fmt"
type TSDataType int8
type TSEncoding uint8
type TSCompressionType uint8
const (
UNKNOWN TSDataType = -1
BOOLEAN TSDataType = 0
INT32 TSDataType = 1
INT64 TSDataType = 2
FLOAT TSDataType = 3
DOUBLE TSDataType = 4
TEXT TSDataType = 5
TIMESTAMP TSDataType = 8
DATE TSDataType = 9
BLOB TSDataType = 10
STRING TSDataType = 11
)
var tsTypeMap = map[string]TSDataType{
"BOOLEAN": BOOLEAN,
"INT32": INT32,
"INT64": INT64,
"FLOAT": FLOAT,
"DOUBLE": DOUBLE,
"TEXT": TEXT,
"TIMESTAMP": TIMESTAMP,
"DATE": DATE,
"BLOB": BLOB,
"STRING": STRING,
}
var byteToTsDataType = map[byte]TSDataType{
0: BOOLEAN,
1: INT32,
2: INT64,
3: FLOAT,
4: DOUBLE,
5: TEXT,
8: TIMESTAMP,
9: DATE,
10: BLOB,
11: STRING,
}
func GetDataTypeByStr(name string) (TSDataType, error) {
dataType, exists := tsTypeMap[name]
if !exists {
return UNKNOWN, fmt.Errorf("invalid input: %v", name)
}
return dataType, nil
}
func getDataTypeByByte(b byte) (TSDataType, error) {
dataType, exists := byteToTsDataType[b]
if !exists {
return UNKNOWN, fmt.Errorf("invalid input: %v", b)
}
return dataType, nil
}
const (
PLAIN TSEncoding = 0
DICTIONARY TSEncoding = 1
RLE TSEncoding = 2
DIFF TSEncoding = 3
TS_2DIFF TSEncoding = 4
BITMAP TSEncoding = 5
GORILLA_V1 TSEncoding = 6
REGULAR TSEncoding = 7
GORILLA TSEncoding = 8
ZIGZAG TSEncoding = 9
FREQ TSEncoding = 10
CHIMP TSEncoding = 11
SPRINTZ TSEncoding = 12
RLBE TSEncoding = 13
)
const (
UNCOMPRESSED TSCompressionType = 0
SNAPPY TSCompressionType = 1
GZIP TSCompressionType = 2
LZ4 TSCompressionType = 7
ZSTD TSCompressionType = 8
LZMA2 TSCompressionType = 9
)
// TSStatusCode
const (
SuccessStatus int32 = 200
IncompatibleVersion int32 = 201
ConfigurationError int32 = 202
StartUpError int32 = 203
ShutDownError int32 = 204
UnsupportedOperation int32 = 300
ExecuteStatementError int32 = 301
MultipleError int32 = 302
IllegalParameter int32 = 303
OverlapWithExistingTask int32 = 304
InternalServerError int32 = 305
RedirectionRecommend int32 = 400
DatabaseNotExist int32 = 500
DatabaseAlreadyExists int32 = 501
SeriesOverflow int32 = 502
TimeseriesAlreadyExist int32 = 503
TimeseriesInBlackList int32 = 504
AliasAlreadyExist int32 = 505
PathAlreadyExist int32 = 506
MetadataError int32 = 507
PathNotExist int32 = 508
IllegalPath int32 = 509
CreateTemplateError int32 = 510
DuplicatedTemplate int32 = 511
UndefinedTemplate int32 = 512
TemplateNotSet int32 = 513
DifferentTemplate int32 = 514
TemplateIsInUse int32 = 515
TemplateIncompatible int32 = 516
SegmentNotFound int32 = 517
PageOutOfSpace int32 = 518
RecordDuplicated int32 = 519
SegmentOutOfSpace int32 = 520
SchemaFileNotExists int32 = 521
OversizeRecord int32 = 522
SchemaFileRedoLogBroken int32 = 523
TemplateNotActivated int32 = 524
SystemReadOnly int32 = 600
StorageEngineError int32 = 601
StorageEngineNotReady int32 = 602
DataregionProcessError int32 = 603
TsfileProcessorError int32 = 604
WriteProcessError int32 = 605
WriteProcessReject int32 = 606
OutOfTtl int32 = 607
CompactionError int32 = 608
AlignedTimeseriesError int32 = 609
WalError int32 = 610
DiskSpaceInsufficient int32 = 611
SqlParseError int32 = 700
SemanticError int32 = 701
GenerateTimeZoneError int32 = 702
SetTimeZoneError int32 = 703
QueryNotAllowed int32 = 704
LogicalOperatorError int32 = 705
LogicalOptimizeError int32 = 706
UnsupportedFillType int32 = 707
QueryProcessError int32 = 708
MppMemoryNotEnough int32 = 709
CloseOperationError int32 = 710
TsblockSerializeError int32 = 711
InternalRequestTimeOut int32 = 712
InternalRequestRetryError int32 = 713
InitAuthError int32 = 800
WrongLoginPassword int32 = 801
NotLogin int32 = 802
NoPermission int32 = 803
UserNotExist int32 = 804
UserAlreadyExist int32 = 805
UserAlreadyHasRole int32 = 806
UserNotHasRole int32 = 807
RoleNotExist int32 = 808
RoleAlreadyExist int32 = 809
AlreadyHasPrivilege int32 = 810
NotHasPrivilege int32 = 811
ClearPermissionCacheError int32 = 812
UnknownAuthPrivilege int32 = 813
UnsupportedAuthOperation int32 = 814
AuthIoException int32 = 815
MigrateRegionError int32 = 900
CreateRegionError int32 = 901
DeleteRegionError int32 = 902
PartitionCacheUpdateError int32 = 903
ConsensusNotInitialized int32 = 904
RegionLeaderChangeError int32 = 905
NoAvailableRegionGroup int32 = 906
DatanodeAlreadyRegistered int32 = 1000
NoEnoughDatanode int32 = 1001
AddConfignodeError int32 = 1002
RemoveConfignodeError int32 = 1003
DatanodeNotExist int32 = 1004
DatanodeStopError int32 = 1005
RemoveDatanodeError int32 = 1006
RegisterRemovedDatanode int32 = 1007
CanNotConnectDatanode int32 = 1008
LoadFileError int32 = 1100
LoadPieceOfTsfileError int32 = 1101
DeserializePieceOfTsfileError int32 = 1102
SyncConnectionError int32 = 1103
SyncFileRedirectionError int32 = 1104
SyncFileError int32 = 1105
CreatePipeSinkError int32 = 1106
PipeError int32 = 1107
PipeserverError int32 = 1108
VerifyMetadataError int32 = 1109
UdfLoadClassError int32 = 1200
UdfDownloadError int32 = 1201
CreateUdfOnDatanodeError int32 = 1202
DropUdfOnDatanodeError int32 = 1203
CreateTriggerError int32 = 1300
DropTriggerError int32 = 1301
TriggerFireError int32 = 1302
TriggerLoadClassError int32 = 1303
TriggerDownloadError int32 = 1304
CreateTriggerInstanceError int32 = 1305
ActiveTriggerInstanceError int32 = 1306
DropTriggerInstanceError int32 = 1307
UpdateTriggerLocationError int32 = 1308
NoSuchCq int32 = 1400
CqAlreadyActive int32 = 1401
CqAlreadyExist int32 = 1402
CqUpdateLastExecTimeError int32 = 1403
)
const (
TimestampColumnName = "Time"
)
agents/go-common/iotdb-client-go/client/rowrecord.go
+33
View File
@@ -0,0 +1,33 @@
/*
* 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
type RowRecord struct {
timestamp int64
fields []*Field
}
func (r *RowRecord) GetFields() []*Field {
return r.fields
}
func (r *RowRecord) GetTimestamp() int64 {
return r.timestamp
}
agents/go-common/iotdb-client-go/client/rpcdataset.go
+829
View File
@@ -0,0 +1,829 @@
package client
import (
"context"
"fmt"
"strconv"
"time"
"github.com/apache/iotdb-client-go/v2/common"
"github.com/apache/iotdb-client-go/v2/rpc"
)
const startIndex = int32(2)
type IoTDBRpcDataSet struct {
sql string
isClosed bool
client *rpc.IClientRPCServiceClient
columnNameList []string // no deduplication
columnTypeList []string // no deduplication
columnOrdinalMap map[string]int32 // used because the server returns deduplicated columns
columnName2TsBlockColumnIndexMap map[string]int32
// column index -> TsBlock column index
columnIndex2TsBlockColumnIndexList []int32
dataTypeForTsBlockColumn []TSDataType
fetchSize int32
timeout *int64
hasCachedRecord bool
lastReadWasNull bool
columnSize int32
sessionId int64
queryId int64
statementId int64
time int64
ignoreTimestamp bool
// indicates that there is still more data in server side and we can call fetchResult to get more
moreData bool
queryResult [][]byte
curTsBlock *TsBlock
queryResultSize int32 // the length of queryResult
queryResultIndex int32 // the index of bytebuffer in queryResult
tsBlockSize int32 // the size of current tsBlock
tsBlockIndex int32 // the row index in current tsBlock
zoneId *time.Location
timeFormat string
timeFactor int32
timePrecision string
}
func NewIoTDBRpcDataSet(sql string, columnNameList []string, columnTypeList []string, columnNameIndex map[string]int32, ignoreTimestamp bool, moreData bool, queryId int64, statementId int64, client *rpc.IClientRPCServiceClient, sessionId int64, queryResult [][]byte, fetchSize int32, timeout *int64, zoneId string, timeFormat string, timeFactor int32, columnIndex2TsBlockColumnIndexList []int32) (rpcDataSet *IoTDBRpcDataSet, err error) {
ds := &IoTDBRpcDataSet{
sessionId: sessionId,
statementId: statementId,
ignoreTimestamp: ignoreTimestamp,
sql: sql,
queryId: queryId,
client: client,
fetchSize: fetchSize,
timeout: timeout,
moreData: moreData,
columnSize: int32(len(columnNameList)),
columnNameList: make([]string, 0, len(columnNameList)+1),
columnTypeList: make([]string, 0, len(columnTypeList)+1),
columnOrdinalMap: make(map[string]int32),
columnName2TsBlockColumnIndexMap: make(map[string]int32),
}
columnStartIndex := int32(1)
resultSetColumnSize := int32(len(columnNameList))
// newly generated or updated columnIndex2TsBlockColumnIndexList.size() may not be equal to
// columnNameList.size()
// so we need startIndexForColumnIndex2TsBlockColumnIndexList to adjust the mapping relation
startIndexForColumnIndex2TsBlockColumnIndexList := int32(0)
// for Time Column in tree model which should always be the first column and its index for
// TsBlockColumn is -1
if !ignoreTimestamp {
ds.columnNameList = append(ds.columnNameList, TimestampColumnName)
ds.columnTypeList = append(ds.columnTypeList, "INT64")
ds.columnName2TsBlockColumnIndexMap[TimestampColumnName] = -1
ds.columnOrdinalMap[TimestampColumnName] = 1
if columnIndex2TsBlockColumnIndexList != nil {
newColumnIndex2TsBlockColumnIndexList := make([]int32, 0, len(columnIndex2TsBlockColumnIndexList)+1)
newColumnIndex2TsBlockColumnIndexList = append(newColumnIndex2TsBlockColumnIndexList, -1)
newColumnIndex2TsBlockColumnIndexList = append(newColumnIndex2TsBlockColumnIndexList, columnIndex2TsBlockColumnIndexList...)
columnIndex2TsBlockColumnIndexList = newColumnIndex2TsBlockColumnIndexList
startIndexForColumnIndex2TsBlockColumnIndexList = 1
}
columnStartIndex += 1
resultSetColumnSize += 1
}
ds.columnNameList = append(ds.columnNameList, columnNameList...)
ds.columnTypeList = append(ds.columnTypeList, columnTypeList...)
if columnIndex2TsBlockColumnIndexList == nil {
columnIndex2TsBlockColumnIndexList = make([]int32, 0, resultSetColumnSize)
if !ignoreTimestamp {
startIndexForColumnIndex2TsBlockColumnIndexList = 1
columnIndex2TsBlockColumnIndexList = append(columnIndex2TsBlockColumnIndexList, -1)
}
// IoTDB 1.3.x servers send columnNameIndex (column name -> deduplicated
// value column index) instead of the ordered index list, because their
// aggregate TsBlock column order may differ from the SELECT list. Prefer
// that mapping so values stay aligned with column names.
for i, columnName := range columnNameList {
if index, ok := columnNameIndex[columnName]; ok {
columnIndex2TsBlockColumnIndexList = append(columnIndex2TsBlockColumnIndexList, index)
} else {
columnIndex2TsBlockColumnIndexList = append(columnIndex2TsBlockColumnIndexList, int32(i))
}
}
}
tsBlockColumnSize := int32(0)
for _, value := range columnIndex2TsBlockColumnIndexList {
if value > tsBlockColumnSize {
tsBlockColumnSize = value
}
}
tsBlockColumnSize += 1
ds.dataTypeForTsBlockColumn = make([]TSDataType, tsBlockColumnSize)
for i, columnName := range columnNameList {
tsBlockColumnIndex := columnIndex2TsBlockColumnIndexList[startIndexForColumnIndex2TsBlockColumnIndexList+int32(i)]
if tsBlockColumnIndex != -1 {
columnType, err := GetDataTypeByStr(columnTypeList[i])
if err != nil {
return nil, err
}
ds.dataTypeForTsBlockColumn[tsBlockColumnIndex] = columnType
}
if _, exists := ds.columnName2TsBlockColumnIndexMap[columnName]; !exists {
ds.columnOrdinalMap[columnName] = int32(i) + columnStartIndex
ds.columnName2TsBlockColumnIndexMap[columnName] = tsBlockColumnIndex
}
}
ds.queryResult = queryResult
if queryResult != nil {
ds.queryResultSize = int32(len(queryResult))
} else {
ds.queryResultSize = 0
}
ds.queryResultIndex = 0
ds.tsBlockSize = 0
ds.tsBlockIndex = -1
if ds.zoneId, err = time.LoadLocation(zoneId); err != nil {
return nil, err
}
ds.timeFormat = timeFormat
ds.timeFactor = timeFactor
if ds.timePrecision, err = getTimePrecision(timeFactor); err != nil {
return nil, err
}
if len(columnIndex2TsBlockColumnIndexList) != len(ds.columnNameList) {
return nil, fmt.Errorf("size of columnIndex2TsBlockColumnIndexList %v doesn't equal to size of columnNameList %v", len(columnIndex2TsBlockColumnIndexList), len(ds.columnNameList))
}
ds.columnIndex2TsBlockColumnIndexList = columnIndex2TsBlockColumnIndexList
return ds, nil
}
func (s *IoTDBRpcDataSet) Close() (err error) {
if s.isClosed {
return nil
}
closeRequest := &rpc.TSCloseOperationReq{
SessionId: s.sessionId,
StatementId: &s.statementId,
QueryId: &s.queryId,
}
var status *common.TSStatus
status, err = s.client.CloseOperation(context.Background(), closeRequest)
if err == nil {
err = VerifySuccess(status)
}
s.client = nil
s.isClosed = true
return err
}
func (s *IoTDBRpcDataSet) Next() (result bool, err error) {
if s.hasCachedBlock() {
s.lastReadWasNull = false
err = s.constructOneRow()
return true, err
}
if s.hasCachedByteBuffer() {
if err = s.constructOneTsBlock(); err != nil {
return false, err
}
err = s.constructOneRow()
return true, err
}
if s.moreData {
hasResultSet, err := s.fetchResults()
if err != nil {
return false, err
}
if hasResultSet && s.hasCachedByteBuffer() {
if err = s.constructOneTsBlock(); err != nil {
return false, err
}
err = s.constructOneRow()
return true, err
}
}
err = s.Close()
if err != nil {
return false, err
}
return false, nil
}
func (s *IoTDBRpcDataSet) fetchResults() (bool, error) {
if s.isClosed {
return false, fmt.Errorf("this data set is already closed")
}
req := rpc.TSFetchResultsReq{
SessionId: s.sessionId,
Statement: s.sql,
FetchSize: s.fetchSize,
QueryId: s.queryId,
IsAlign: true,
}
req.Timeout = s.timeout
resp, err := s.client.FetchResultsV2(context.Background(), &req)
if err != nil {
return false, err
}
if err = VerifySuccess(resp.Status); err != nil {
return false, err
}
s.moreData = resp.GetMoreData()
if !resp.HasResultSet {
err = s.Close()
} else {
s.queryResult = resp.GetQueryResult_()
s.queryResultIndex = 0
if s.queryResult != nil {
s.queryResultSize = int32(len(s.queryResult))
} else {
s.queryResultSize = 0
}
s.tsBlockSize = 0
s.tsBlockIndex = -1
}
return resp.HasResultSet, err
}
func (s *IoTDBRpcDataSet) hasCachedBlock() bool {
return s.curTsBlock != nil && s.tsBlockIndex < s.tsBlockSize-1
}
func (s *IoTDBRpcDataSet) hasCachedByteBuffer() bool {
return s.queryResult != nil && s.queryResultIndex < s.queryResultSize
}
func (s *IoTDBRpcDataSet) constructOneRow() (err error) {
s.tsBlockIndex++
s.hasCachedRecord = true
s.time, err = s.curTsBlock.GetTimeColumn().GetLong(s.tsBlockIndex)
return err
}
func (s *IoTDBRpcDataSet) constructOneTsBlock() (err error) {
s.lastReadWasNull = false
curTsBlockBytes := s.queryResult[s.queryResultIndex]
s.queryResultIndex = s.queryResultIndex + 1
s.curTsBlock, err = DeserializeTsBlock(curTsBlockBytes)
if err != nil {
return err
}
s.tsBlockIndex = -1
s.tsBlockSize = s.curTsBlock.GetPositionCount()
return nil
}
func (s *IoTDBRpcDataSet) isNullByIndex(columnIndex int32) (bool, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return false, err
} else {
return s.isNull(index, s.tsBlockIndex), nil
}
}
func (s *IoTDBRpcDataSet) isNullByColumnName(columnName string) (bool, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return false, err
} else {
return s.isNull(index, s.tsBlockIndex), nil
}
}
func (s *IoTDBRpcDataSet) isNull(index int32, rowNum int32) bool {
// -1 for time column which will never be null
return index >= 0 && s.curTsBlock.GetColumn(index).IsNull(rowNum)
}
func (s *IoTDBRpcDataSet) getBooleanByIndex(columnIndex int32) (bool, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return false, err
} else {
return s.getBooleanByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getBoolean(columnName string) (bool, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return false, err
} else {
return s.getBooleanByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getBooleanByTsBlockColumnIndex(tsBlockColumnIndex int32) (bool, error) {
if err := s.checkRecord(); err != nil {
return false, err
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBoolean(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return false, nil
}
}
func (s *IoTDBRpcDataSet) getDoubleByIndex(columnIndex int32) (float64, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return 0, err
} else {
return s.getDoubleByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getDouble(columnName string) (float64, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return 0, err
} else {
return s.getDoubleByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getDoubleByTsBlockColumnIndex(tsBlockColumnIndex int32) (float64, error) {
if err := s.checkRecord(); err != nil {
return 0, err
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetDouble(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return 0, nil
}
}
func (s *IoTDBRpcDataSet) getFloatByIndex(columnIndex int32) (float32, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return 0, err
} else {
return s.getFloatByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getFloat(columnName string) (float32, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return 0, err
} else {
return s.getFloatByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getFloatByTsBlockColumnIndex(tsBlockColumnIndex int32) (float32, error) {
if err := s.checkRecord(); err != nil {
return 0, err
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetFloat(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return 0, nil
}
}
func (s *IoTDBRpcDataSet) getIntByIndex(columnIndex int32) (int32, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return 0, err
} else {
return s.getIntByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getInt(columnName string) (int32, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return 0, err
} else {
return s.getIntByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getIntByTsBlockColumnIndex(tsBlockColumnIndex int32) (int32, error) {
if err := s.checkRecord(); err != nil {
return 0, err
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
dataType := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetDataType()
if dataType == INT64 {
if v, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetLong(s.tsBlockIndex); err != nil {
return 0, err
} else {
return int32(v), nil
}
}
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetInt(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return 0, nil
}
}
func (s *IoTDBRpcDataSet) getLongByIndex(columnIndex int32) (int64, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return 0, err
} else {
return s.getLongByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getLong(columnName string) (int64, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return 0, err
} else {
return s.getLongByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getLongByTsBlockColumnIndex(tsBlockColumnIndex int32) (int64, error) {
if err := s.checkRecord(); err != nil {
return 0, err
}
if tsBlockColumnIndex < 0 {
s.lastReadWasNull = false
return s.curTsBlock.GetTimeByIndex(s.tsBlockIndex)
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
dataType := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetDataType()
if dataType == INT32 {
if v, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetInt(s.tsBlockIndex); err != nil {
return 0, err
} else {
return int64(v), nil
}
}
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetLong(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return 0, nil
}
}
func (s *IoTDBRpcDataSet) getBinaryByIndex(columnIndex int32) (*Binary, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return nil, err
} else {
return s.getBinaryByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getBinary(columnName string) (*Binary, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return nil, err
} else {
return s.getBinaryByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getBinaryByTsBlockColumnIndex(tsBlockColumnIndex int32) (*Binary, error) {
if err := s.checkRecord(); err != nil {
return nil, err
}
if !s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = false
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBinary(s.tsBlockIndex)
} else {
s.lastReadWasNull = true
return nil, nil
}
}
func (s *IoTDBRpcDataSet) getObjectByIndex(columnIndex int32) (interface{}, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return nil, err
} else {
return s.getObjectByTsBlockIndex(index)
}
}
func (s *IoTDBRpcDataSet) getObject(columnName string) (interface{}, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return nil, err
} else {
return s.getObjectByTsBlockIndex(index)
}
}
func (s *IoTDBRpcDataSet) getObjectByTsBlockIndex(tsBlockColumnIndex int32) (interface{}, error) {
if err := s.checkRecord(); err != nil {
return nil, err
}
if s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = true
return nil, nil
}
s.lastReadWasNull = false
dataType := s.getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex)
switch dataType {
case BOOLEAN, INT32, INT64, FLOAT, DOUBLE:
return s.curTsBlock.GetColumn(tsBlockColumnIndex).GetObject(s.tsBlockIndex)
case TIMESTAMP:
var timestamp int64
var err error
if tsBlockColumnIndex == -1 {
timestamp, err = s.curTsBlock.GetTimeByIndex(s.tsBlockIndex)
} else {
timestamp, err = s.curTsBlock.GetColumn(tsBlockColumnIndex).GetLong(s.tsBlockIndex)
}
if err != nil {
return nil, err
}
return convertToTimestamp(timestamp, s.timeFactor), nil
case TEXT, STRING:
if binary, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBinary(s.tsBlockIndex); err != nil {
return nil, err
} else {
return binary.GetStringValue(), nil
}
case BLOB:
if binary, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetBinary(s.tsBlockIndex); err != nil {
return nil, err
} else {
return binary.GetValues(), nil
}
case DATE:
if value, err := s.curTsBlock.GetColumn(tsBlockColumnIndex).GetInt(s.tsBlockIndex); err != nil {
return nil, err
} else {
return Int32ToDate(value)
}
default:
return nil, nil
}
}
func (s *IoTDBRpcDataSet) getStringByIndex(columnIndex int32) (string, error) {
columnIndex, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex)
if err != nil {
return "", err
}
return s.getStringByTsBlockColumnIndex(columnIndex)
}
func (s *IoTDBRpcDataSet) getString(columnName string) (string, error) {
columnIndex, err := s.getTsBlockColumnIndexForColumnName(columnName)
if err != nil {
return "", err
}
return s.getStringByTsBlockColumnIndex(columnIndex)
}
func (s *IoTDBRpcDataSet) getStringByTsBlockColumnIndex(tsBlockColumnIndex int32) (string, error) {
if err := s.checkRecord(); err != nil {
return "", err
}
// to keep compatibility, tree model should return a long value for time column
if tsBlockColumnIndex == -1 {
timestamp, err := s.curTsBlock.GetTimeByIndex(s.tsBlockIndex)
if err != nil {
return "", err
}
return int64ToString(timestamp), nil
}
if s.isNull(tsBlockColumnIndex, s.tsBlockIndex) {
s.lastReadWasNull = true
return "", nil
}
s.lastReadWasNull = false
return s.getStringByTsBlockColumnIndexAndDataType(tsBlockColumnIndex, s.getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex))
}
func (s *IoTDBRpcDataSet) getStringByTsBlockColumnIndexAndDataType(index int32, tsDataType TSDataType) (string, error) {
switch tsDataType {
case BOOLEAN:
if v, err := s.curTsBlock.GetColumn(index).GetBoolean(s.tsBlockIndex); err != nil {
return "", nil
} else {
return strconv.FormatBool(v), nil
}
case INT32:
if v, err := s.curTsBlock.GetColumn(index).GetInt(s.tsBlockIndex); err != nil {
return "", err
} else {
return int32ToString(v), nil
}
case INT64:
if v, err := s.curTsBlock.GetColumn(index).GetLong(s.tsBlockIndex); err != nil {
return "", err
} else {
return int64ToString(v), nil
}
case TIMESTAMP:
if v, err := s.curTsBlock.GetColumn(index).GetLong(s.tsBlockIndex); err != nil {
return "", err
} else {
return formatDatetime(DEFAULT_TIME_FORMAT, s.timePrecision, v, s.zoneId), nil
}
case FLOAT:
if v, err := s.curTsBlock.GetColumn(index).GetFloat(s.tsBlockIndex); err != nil {
return "", err
} else {
return float32ToString(v), nil
}
case DOUBLE:
if v, err := s.curTsBlock.GetColumn(index).GetDouble(s.tsBlockIndex); err != nil {
return "", err
} else {
return float64ToString(v), nil
}
case TEXT, STRING:
if v, err := s.curTsBlock.GetColumn(index).GetBinary(s.tsBlockIndex); err != nil {
return "", err
} else {
return v.GetStringValue(), nil
}
case BLOB:
if v, err := s.curTsBlock.GetColumn(index).GetBinary(s.tsBlockIndex); err != nil {
return "", err
} else {
return bytesToHexString(v.values), nil
}
case DATE:
v, err := s.curTsBlock.GetColumn(index).GetInt(s.tsBlockIndex)
if err != nil {
return "", err
}
t, err := Int32ToDate(v)
if err != nil {
return "", err
}
return t.Format("2006-01-02"), nil
}
return "", nil
}
func (s *IoTDBRpcDataSet) getTimestampByIndex(columnIndex int32) (time.Time, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return time.Time{}, err
} else {
return s.getTimestampByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getTimestamp(columnName string) (time.Time, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return time.Time{}, err
} else {
return s.getTimestampByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) getTimestampByTsBlockColumnIndex(tsBlockColumnIndex int32) (time.Time, error) {
if value, err := s.getLongByTsBlockColumnIndex(tsBlockColumnIndex); err != nil {
return time.Time{}, err
} else {
return convertToTimestamp(value, s.timeFactor), nil
}
}
func (s *IoTDBRpcDataSet) GetDateByIndex(columnIndex int32) (time.Time, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return time.Time{}, err
} else {
return s.GetDateByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) GetDate(columnName string) (time.Time, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return time.Time{}, err
} else {
return s.GetDateByTsBlockColumnIndex(index)
}
}
func (s *IoTDBRpcDataSet) GetDateByTsBlockColumnIndex(tsBlockColumnIndex int32) (time.Time, error) {
if value, err := s.getIntByTsBlockColumnIndex(tsBlockColumnIndex); err != nil {
return time.Time{}, err
} else {
return Int32ToDate(value)
}
}
func (s *IoTDBRpcDataSet) getDataTypeByIndex(columnIndex int32) (TSDataType, error) {
if index, err := s.getTsBlockColumnIndexForColumnIndex(columnIndex); err != nil {
return UNKNOWN, err
} else {
return s.getDataTypeByTsBlockColumnIndex(index), nil
}
}
func (s *IoTDBRpcDataSet) getDataType(columnName string) (TSDataType, error) {
if index, err := s.getTsBlockColumnIndexForColumnName(columnName); err != nil {
return UNKNOWN, err
} else {
return s.getDataTypeByTsBlockColumnIndex(index), nil
}
}
func (s *IoTDBRpcDataSet) getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex int32) TSDataType {
if tsBlockColumnIndex < 0 {
return TIMESTAMP
} else {
return s.dataTypeForTsBlockColumn[tsBlockColumnIndex]
}
}
func (s *IoTDBRpcDataSet) findColumn(columnName string) int32 {
return s.columnOrdinalMap[columnName]
}
func (s *IoTDBRpcDataSet) findColumnNameByIndex(columnIndex int32) (string, error) {
if columnIndex <= 0 {
return "", fmt.Errorf("column index should start from 1")
}
if columnIndex > int32(len(s.columnNameList)) {
return "", fmt.Errorf("column index %d out of range %d", columnIndex, len(s.columnNameList))
}
return s.columnNameList[columnIndex-1], nil
}
// return -1 for time column of tree model
func (s *IoTDBRpcDataSet) getTsBlockColumnIndexForColumnName(columnName string) (int32, error) {
if index, exists := s.columnName2TsBlockColumnIndexMap[columnName]; !exists {
return 0, fmt.Errorf("unknown column name: %v", columnName)
} else {
return index, nil
}
}
func (s *IoTDBRpcDataSet) getTsBlockColumnIndexForColumnIndex(columnIndex int32) (int32, error) {
if columnIndex-1 >= int32(len(s.columnIndex2TsBlockColumnIndexList)) || int32(columnIndex-1) < 0 {
return 0, fmt.Errorf("index %v out of range %v", columnIndex-1, len(s.columnIndex2TsBlockColumnIndexList))
}
return s.columnIndex2TsBlockColumnIndexList[columnIndex-1], nil
}
func (s *IoTDBRpcDataSet) checkRecord() (err error) {
if s.queryResultIndex > s.queryResultSize || s.tsBlockIndex >= s.tsBlockSize || s.queryResult == nil || s.curTsBlock == nil {
err = fmt.Errorf("no record remains")
}
return err
}
func (s *IoTDBRpcDataSet) GetValueColumnStartIndex() int32 {
if s.ignoreTimestamp {
return 0
} else {
return 1
}
}
func (s *IoTDBRpcDataSet) GetColumnSize() int32 {
return int32(len(s.columnNameList))
}
func (s *IoTDBRpcDataSet) GetColumnTypeList() []string {
return s.columnTypeList
}
func (s *IoTDBRpcDataSet) GetColumnNameTypeList() []string {
return s.columnTypeList
}
func (s *IoTDBRpcDataSet) IsClosed() bool {
return s.isClosed
}
func (s *IoTDBRpcDataSet) GetFetchSize() int32 {
return s.fetchSize
}
func (s *IoTDBRpcDataSet) SetFetchSize(fetchSize int32) {
s.fetchSize = fetchSize
}
func (s *IoTDBRpcDataSet) HasCachedRecord() bool {
return s.hasCachedRecord
}
func (s *IoTDBRpcDataSet) SetHasCachedRecord(hasCachedRecord bool) {
s.hasCachedRecord = hasCachedRecord
}
func (s *IoTDBRpcDataSet) IsLastReadWasNull() bool {
return s.lastReadWasNull
}
func (s *IoTDBRpcDataSet) GetCurrentRowTime() int64 {
return s.time
}
func (s *IoTDBRpcDataSet) IsIgnoreTimestamp() bool {
return s.ignoreTimestamp
}
agents/go-common/iotdb-client-go/client/session.go
+1477
View File
File diff suppressed because it is too large Load Diff
agents/go-common/iotdb-client-go/client/sessiondataset.go
+131
View File
@@ -0,0 +1,131 @@
package client
import (
"time"
"github.com/apache/iotdb-client-go/v2/rpc"
)
type SessionDataSet struct {
ioTDBRpcDataSet *IoTDBRpcDataSet
}
func NewSessionDataSet(sql string, columnNameList []string, columnTypeList []string, columnNameIndex map[string]int32, queryId int64, statementId int64, client *rpc.IClientRPCServiceClient, sessionId int64, queryResult [][]byte, ignoreTimestamp bool, timeout *int64, moreData bool, fetchSize int32, zoneId string, timeFactor int32, columnIndex2TsBlockColumnIndexList []int32) (*SessionDataSet, error) {
rpcDataSet, err := NewIoTDBRpcDataSet(sql, columnNameList, columnTypeList, columnNameIndex, ignoreTimestamp, moreData, queryId, statementId, client, sessionId, queryResult, fetchSize, timeout, zoneId, DEFAULT_TIME_FORMAT, timeFactor, columnIndex2TsBlockColumnIndexList)
if err != nil {
return nil, err
}
return &SessionDataSet{ioTDBRpcDataSet: rpcDataSet}, nil
}
func (s *SessionDataSet) Next() (bool, error) {
return s.ioTDBRpcDataSet.Next()
}
func (s *SessionDataSet) Close() error {
return s.ioTDBRpcDataSet.Close()
}
func (s *SessionDataSet) IsNull(columnName string) (bool, error) {
return s.ioTDBRpcDataSet.isNullByColumnName(columnName)
}
func (s *SessionDataSet) IsNullByIndex(columnIndex int32) (bool, error) {
return s.ioTDBRpcDataSet.isNullByIndex(columnIndex)
}
func (s *SessionDataSet) GetBooleanByIndex(columnIndex int32) (bool, error) {
return s.ioTDBRpcDataSet.getBooleanByIndex(columnIndex)
}
func (s *SessionDataSet) GetBoolean(columnName string) (bool, error) {
return s.ioTDBRpcDataSet.getBoolean(columnName)
}
func (s *SessionDataSet) GetDoubleByIndex(columnIndex int32) (float64, error) {
return s.ioTDBRpcDataSet.getDoubleByIndex(columnIndex)
}
func (s *SessionDataSet) GetDouble(columnName string) (float64, error) {
return s.ioTDBRpcDataSet.getDouble(columnName)
}
func (s *SessionDataSet) GetFloatByIndex(columnIndex int32) (float32, error) {
return s.ioTDBRpcDataSet.getFloatByIndex(columnIndex)
}
func (s *SessionDataSet) GetFloat(columnName string) (float32, error) {
return s.ioTDBRpcDataSet.getFloat(columnName)
}
func (s *SessionDataSet) GetIntByIndex(columnIndex int32) (int32, error) {
return s.ioTDBRpcDataSet.getIntByIndex(columnIndex)
}
func (s *SessionDataSet) GetInt(columnName string) (int32, error) {
return s.ioTDBRpcDataSet.getInt(columnName)
}
func (s *SessionDataSet) GetLongByIndex(columnIndex int32) (int64, error) {
return s.ioTDBRpcDataSet.getLongByIndex(columnIndex)
}
func (s *SessionDataSet) GetLong(columnName string) (int64, error) {
return s.ioTDBRpcDataSet.getLong(columnName)
}
func (s *SessionDataSet) GetObjectByIndex(columnIndex int32) (interface{}, error) {
return s.ioTDBRpcDataSet.getObjectByIndex(columnIndex)
}
func (s *SessionDataSet) GetObject(columnName string) (interface{}, error) {
return s.ioTDBRpcDataSet.getObject(columnName)
}
func (s *SessionDataSet) GetStringByIndex(columnIndex int32) (string, error) {
return s.ioTDBRpcDataSet.getStringByIndex(columnIndex)
}
func (s *SessionDataSet) GetString(columnName string) (string, error) {
return s.ioTDBRpcDataSet.getString(columnName)
}
func (s *SessionDataSet) GetTimestampByIndex(columnIndex int32) (time.Time, error) {
return s.ioTDBRpcDataSet.getTimestampByIndex(columnIndex)
}
func (s *SessionDataSet) GetTimestamp(columnName string) (time.Time, error) {
return s.ioTDBRpcDataSet.getTimestamp(columnName)
}
func (s *SessionDataSet) GetDateByIndex(columnIndex int32) (time.Time, error) {
return s.ioTDBRpcDataSet.GetDateByIndex(columnIndex)
}
func (s *SessionDataSet) GetDate(columnName string) (time.Time, error) {
return s.ioTDBRpcDataSet.GetDate(columnName)
}
func (s *SessionDataSet) GetBlobByIndex(columnIndex int32) (*Binary, error) {
return s.ioTDBRpcDataSet.getBinaryByIndex(columnIndex)
}
func (s *SessionDataSet) GetBlob(columnName string) (*Binary, error) {
return s.ioTDBRpcDataSet.getBinary(columnName)
}
func (s *SessionDataSet) FindColumn(columnName string) int32 {
return s.ioTDBRpcDataSet.findColumn(columnName)
}
func (s *SessionDataSet) GetColumnNames() []string {
return s.ioTDBRpcDataSet.columnNameList
}
func (s *SessionDataSet) GetColumnTypes() []string {
return s.ioTDBRpcDataSet.columnTypeList
}
func (s *SessionDataSet) GetCurrentRowTime() time.Time {
return convertToTimestamp(s.ioTDBRpcDataSet.GetCurrentRowTime(), s.ioTDBRpcDataSet.timeFactor)
}
agents/go-common/iotdb-client-go/client/sessionpool.go
+201
View File
@@ -0,0 +1,201 @@
/*
* 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 (
"errors"
"log"
"runtime"
"time"
)
var errTimeout = errors.New("get session timeout")
var errPoolClosed = errors.New("sessionPool has closed")
var defaultMultiple = 5
type SessionPool struct {
config *PoolConfig
maxSize int
waitToGetSessionTimeoutInMs int
enableCompression bool
connectionTimeoutInMs int
ch chan Session
sem chan int8
}
type PoolConfig struct {
Host string
Port string
NodeUrls []string
UserName string
Password string
FetchSize int32
TimeZone string
ConnectRetryMax int
Database string
TLSConfig *TLSConfig
sqlDialect string
}
func NewSessionPool(conf *PoolConfig, maxSize, connectionTimeoutInMs, waitToGetSessionTimeoutInMs int,
enableCompression bool) SessionPool {
return newSessionPoolWithSqlDialect(conf, maxSize, connectionTimeoutInMs, waitToGetSessionTimeoutInMs, enableCompression, TreeSqlDialect)
}
func newSessionPoolWithSqlDialect(conf *PoolConfig, maxSize, connectionTimeoutInMs, waitToGetSessionTimeoutInMs int,
enableCompression bool, sqlDialect string) SessionPool {
conf.sqlDialect = sqlDialect
if maxSize <= 0 {
maxSize = runtime.NumCPU() * defaultMultiple
}
var ch = make(chan Session, maxSize)
var sem = make(chan int8, maxSize)
return SessionPool{
config: conf,
maxSize: maxSize,
waitToGetSessionTimeoutInMs: waitToGetSessionTimeoutInMs,
connectionTimeoutInMs: connectionTimeoutInMs,
enableCompression: enableCompression,
ch: ch,
sem: sem,
}
}
func (spool *SessionPool) GetSession() (session Session, err error) {
for {
select {
case spool.sem <- 1:
select {
case session, ok := <-spool.ch:
if ok {
return session, nil
} else {
log.Println("sessionPool has closed")
return session, errPoolClosed
}
default:
config := spool.config
session, err = spool.ConstructSession(config)
if err != nil {
<-spool.sem
}
return session, err
}
case <-time.After(time.Millisecond * time.Duration(spool.waitToGetSessionTimeoutInMs)):
log.Println("get session timeout")
return session, errTimeout
}
}
}
func (spool *SessionPool) getTableSession() (ITableSession, error) {
tableSession := PooledTableSession{}
session, err := spool.GetSession()
if err != nil {
return nil, err
}
tableSession.session = session
tableSession.sessionPool = spool
return &tableSession, nil
}
func (spool *SessionPool) ConstructSession(config *PoolConfig) (session Session, err error) {
if len(config.NodeUrls) > 0 {
session, err = newClusterSessionWithSqlDialect(getClusterSessionConfig(config))
if err != nil {
return session, err
}
if err = session.OpenCluster(spool.enableCompression); err != nil {
log.Print(err)
return session, err
}
} else {
session = newSessionWithSpecifiedSqlDialect(getSessionConfig(config))
if err := session.Open(spool.enableCompression, spool.connectionTimeoutInMs); err != nil {
log.Print(err)
return session, err
}
}
return session, nil
}
func getSessionConfig(config *PoolConfig) *Config {
return &Config{
Host: config.Host,
Port: config.Port,
UserName: config.UserName,
Password: config.Password,
FetchSize: config.FetchSize,
TimeZone: config.TimeZone,
ConnectRetryMax: config.ConnectRetryMax,
sqlDialect: config.sqlDialect,
Database: config.Database,
TLSConfig: config.TLSConfig,
}
}
func getClusterSessionConfig(config *PoolConfig) *ClusterConfig {
return &ClusterConfig{
NodeUrls: config.NodeUrls,
UserName: config.UserName,
Password: config.Password,
FetchSize: config.FetchSize,
TimeZone: config.TimeZone,
ConnectRetryMax: config.ConnectRetryMax,
sqlDialect: config.sqlDialect,
Database: config.Database,
TLSConfig: config.TLSConfig,
}
}
func (spool *SessionPool) PutBack(session Session) {
defer func() {
if r := recover(); r != nil {
session.Close()
}
}()
if session.trans != nil && session.trans.IsOpen() {
spool.ch <- session
}
<-spool.sem
}
func (spool *SessionPool) dropSession(session Session) {
defer func() {
if e := recover(); e != nil {
if session.trans != nil && session.trans.IsOpen() {
session.Close()
}
}
}()
err := session.Close()
if err != nil {
log.Println("Failed to close session ", session)
}
<-spool.sem
}
func (spool *SessionPool) Close() {
close(spool.ch)
for s := range spool.ch {
s.Close()
}
close(spool.sem)
}
agents/go-common/iotdb-client-go/client/tablesession.go
+159
View File
@@ -0,0 +1,159 @@
/*
* 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
// ITableSession defines an interface for interacting with IoTDB tables.
// It supports operations such as data insertion, executing queries, and closing the session.
// Implementations of this interface are expected to manage connections and ensure
// proper resource cleanup.
//
// Each method may return an error to indicate issues such as connection errors
// or execution failures.
//
// Since this interface includes a Close method, it is recommended to use
// defer to ensure the session is properly closed.
type ITableSession interface {
// Insert inserts a Tablet into the database.
//
// Parameters:
// - tablet: A pointer to a Tablet containing time-series data to be inserted.
//
// Returns:
// - err: An error if an issue occurs during the operation, such as a connection error or execution failure.
Insert(tablet *Tablet) error
// ExecuteNonQueryStatement executes a non-query SQL statement, such as a DDL or DML command.
//
// Parameters:
// - sql: The SQL statement to execute.
//
// Returns:
// - err: An error if an issue occurs during the operation, such as a connection error or execution failure.
ExecuteNonQueryStatement(sql string) error
// ExecuteQueryStatement executes a query SQL statement and returns the result set.
//
// Parameters:
// - sql: The SQL query statement to execute.
// - timeoutInMs: A pointer to the timeout duration in milliseconds for the query execution.
//
// Returns:
// - result: A pointer to SessionDataSet containing the query results.
// - err: An error if an issue occurs during the operation, such as a connection error or execution failure.
ExecuteQueryStatement(sql string, timeoutInMs *int64) (*SessionDataSet, error)
// Close closes the session, releasing any held resources.
//
// Returns:
// - err: An error if there is an issue with closing the IoTDB connection.
Close() (err error)
}
// TableSession represents a session for interacting with IoTDB in relational mode.
// It wraps a Session instance, providing methods for executing SQL statements
// and managing the lifecycle of the session.
type TableSession struct {
session Session
}
// NewTableSession creates a new TableSession instance using the provided configuration.
//
// Parameters:
// - config: The configuration for the session.
// - enableRPCCompression: A boolean indicating whether RPC compression is enabled.
// - connectionTimeoutInMs: The timeout in milliseconds for establishing a connection.
//
// Returns:
// - An ITableSession instance if the session is successfully created.
// - An error if there is an issue during session initialization.
func NewTableSession(config *Config, enableRPCCompression bool, connectionTimeoutInMs int) (ITableSession, error) {
config.sqlDialect = TableSqlDialect
session := newSessionWithSpecifiedSqlDialect(config)
if err := session.Open(enableRPCCompression, connectionTimeoutInMs); err != nil {
return nil, err
}
return &TableSession{session: session}, nil
}
// NewClusterTableSession creates a new TableSession instance for a cluster setup.
//
// Parameters:
// - clusterConfig: The configuration for the cluster session.
// - enableRPCCompression: A boolean indicating whether RPC compression is enabled.
//
// Returns:
// - An ITableSession instance if the session is successfully created.
// - An error if there is an issue during session initialization.
func NewClusterTableSession(clusterConfig *ClusterConfig, enableRPCCompression bool) (ITableSession, error) {
clusterConfig.sqlDialect = TableSqlDialect
session, err := newClusterSessionWithSqlDialect(clusterConfig)
if err != nil {
return nil, err
}
if err = session.OpenCluster(enableRPCCompression); err != nil {
return nil, err
}
return &TableSession{session: session}, nil
}
// Insert inserts a Tablet into the IoTDB.
//
// Parameters:
// - tablet: A pointer to a Tablet containing the data to be inserted.
//
// Returns:
// - err: An error if the operation fails.
func (s *TableSession) Insert(tablet *Tablet) error {
return s.session.insertRelationalTablet(tablet)
}
// ExecuteNonQueryStatement executes a non-query SQL statement, such as a DDL or DML command.
//
// Parameters:
// - sql: The SQL statement to be executed.
//
// Returns:
// - err: An error if the operation fails.
func (s *TableSession) ExecuteNonQueryStatement(sql string) error {
return s.session.ExecuteNonQueryStatement(sql)
}
// ExecuteQueryStatement executes a query SQL statement and retrieves the result set.
//
// Parameters:
// - sql: The SQL query to be executed.
// - timeoutInMs: (Optional) A pointer to the timeout duration in milliseconds for the query.
//
// Returns:
// - result: A pointer to SessionDataSet containing the query results.
// - err: An error if the operation fails.
func (s *TableSession) ExecuteQueryStatement(sql string, timeoutInMs *int64) (*SessionDataSet, error) {
return s.session.ExecuteQueryStatement(sql, timeoutInMs)
}
// Close closes the TableSession, releasing any associated resources.
//
// Returns:
// - err: An error if the session fails to close properly.
func (s *TableSession) Close() error {
return s.session.Close()
}
agents/go-common/iotdb-client-go/client/tablesessionpool.go
+189
View File
@@ -0,0 +1,189 @@
/*
* 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 (
"errors"
"log"
"sync/atomic"
)
var ErrTableSessionClosed = errors.New("table session has been closed")
// TableSessionPool manages a pool of ITableSession instances, enabling efficient
// reuse and management of resources. It provides methods to acquire a session
// from the pool and to close the pool, releasing all held resources.
//
// This implementation ensures proper lifecycle management of sessions,
// including efficient reuse and cleanup of resources.
type TableSessionPool struct {
sessionPool SessionPool
}
// NewTableSessionPool creates a new TableSessionPool with the specified configuration.
//
// Parameters:
// - conf: PoolConfig defining the configuration for the pool.
// - maxSize: The maximum number of sessions the pool can hold.
// - connectionTimeoutInMs: Timeout for establishing a connection in milliseconds.
// - waitToGetSessionTimeoutInMs: Timeout for waiting to acquire a session in milliseconds.
// - enableCompression: A boolean indicating whether to enable compression.
//
// Returns:
// - A TableSessionPool instance.
func NewTableSessionPool(conf *PoolConfig, maxSize, connectionTimeoutInMs, waitToGetSessionTimeoutInMs int,
enableCompression bool) TableSessionPool {
return TableSessionPool{sessionPool: newSessionPoolWithSqlDialect(conf, maxSize, connectionTimeoutInMs, waitToGetSessionTimeoutInMs, enableCompression, TableSqlDialect)}
}
// GetSession acquires an ITableSession instance from the pool.
//
// Returns:
// - A usable ITableSession instance for interacting with IoTDB.
// - An error if a session cannot be acquired.
func (spool *TableSessionPool) GetSession() (ITableSession, error) {
return spool.sessionPool.getTableSession()
}
// Close closes the TableSessionPool, releasing all held resources.
// Once closed, no further sessions can be acquired from the pool.
func (spool *TableSessionPool) Close() {
spool.sessionPool.Close()
}
// PooledTableSession represents a session managed within a TableSessionPool.
// It ensures proper cleanup and reusability of the session.
type PooledTableSession struct {
session Session
sessionPool *SessionPool
closed int32
}
// isConnectionError returns true if the error is a connection-level error
// (i.e., not a server-side execution error indicated by TSStatus).
func isConnectionError(err error) bool {
if err == nil {
return false
}
var exeErr *ExecutionError
if errors.As(err, &exeErr) {
return false
}
var batchErr *BatchError
if errors.As(err, &batchErr) {
return false
}
return true
}
// Insert inserts a Tablet into the database.
//
// Parameters:
// - tablet: A pointer to a Tablet containing time-series data to be inserted.
//
// Returns:
// - err: An error if an issue occurs during the operation.
func (s *PooledTableSession) Insert(tablet *Tablet) error {
if atomic.LoadInt32(&s.closed) == 1 {
return ErrTableSessionClosed
}
err := s.session.insertRelationalTablet(tablet)
if err == nil {
return nil
}
if isConnectionError(err) {
if atomic.CompareAndSwapInt32(&s.closed, 0, 1) {
s.sessionPool.dropSession(s.session)
s.session = Session{}
}
}
return err
}
// ExecuteNonQueryStatement executes a non-query SQL statement, such as a DDL or DML command.
//
// Parameters:
// - sql: The SQL statement to execute.
//
// Returns:
// - err: An error if an issue occurs during the operation.
func (s *PooledTableSession) ExecuteNonQueryStatement(sql string) error {
if atomic.LoadInt32(&s.closed) == 1 {
return ErrTableSessionClosed
}
err := s.session.ExecuteNonQueryStatement(sql)
if err == nil {
return nil
}
if isConnectionError(err) {
if atomic.CompareAndSwapInt32(&s.closed, 0, 1) {
s.sessionPool.dropSession(s.session)
s.session = Session{}
}
}
return err
}
// ExecuteQueryStatement executes a query SQL statement and returns the result set.
//
// Parameters:
// - sql: The SQL query statement to execute.
// - timeoutInMs: A pointer to the timeout duration in milliseconds for query execution.
//
// Returns:
// - result: A pointer to SessionDataSet containing the query results.
// - err: An error if an issue occurs during the operation.
func (s *PooledTableSession) ExecuteQueryStatement(sql string, timeoutInMs *int64) (*SessionDataSet, error) {
if atomic.LoadInt32(&s.closed) == 1 {
return nil, ErrTableSessionClosed
}
sessionDataSet, err := s.session.ExecuteQueryStatement(sql, timeoutInMs)
if err == nil {
return sessionDataSet, nil
}
if isConnectionError(err) {
if atomic.CompareAndSwapInt32(&s.closed, 0, 1) {
s.sessionPool.dropSession(s.session)
s.session = Session{}
}
}
return nil, err
}
// Close closes the PooledTableSession, releasing it back to the pool.
//
// Returns:
// - err: An error if there is an issue with session closure or cleanup.
func (s *PooledTableSession) Close() error {
if atomic.CompareAndSwapInt32(&s.closed, 0, 1) {
if s.session.config.Database != s.sessionPool.config.Database && s.sessionPool.config.Database != "" {
err := s.session.ExecuteNonQueryStatement("use " + s.sessionPool.config.Database)
if err != nil {
log.Println("Failed to change back database by executing: use ", s.sessionPool.config.Database)
s.sessionPool.dropSession(s.session)
s.session = Session{}
return nil
}
}
s.sessionPool.PutBack(s.session)
s.session = Session{}
}
return nil
}
agents/go-common/iotdb-client-go/client/tablet.go
+384
View File
@@ -0,0 +1,384 @@
/*
* 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 (
"bytes"
"encoding/binary"
"fmt"
"reflect"
"sort"
"time"
)
type MeasurementSchema struct {
Measurement string
DataType TSDataType
}
type ColumnCategory int8
const (
TAG = iota
FIELD
ATTRIBUTE
)
type Tablet struct {
insertTargetName string
measurementSchemas []*MeasurementSchema
columnCategories []ColumnCategory
timestamps []int64
values []interface{}
bitMaps []*BitMap
maxRowNumber int
RowSize int
}
func (t *Tablet) Len() int {
return t.RowSize
}
func (t *Tablet) Swap(i, j int) {
for index, schema := range t.measurementSchemas {
switch schema.DataType {
case BOOLEAN:
sortedSlice := t.values[index].([]bool)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
case INT32, DATE:
sortedSlice := t.values[index].([]int32)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
case INT64, TIMESTAMP:
sortedSlice := t.values[index].([]int64)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
case FLOAT:
sortedSlice := t.values[index].([]float32)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
case DOUBLE:
sortedSlice := t.values[index].([]float64)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
case TEXT, BLOB, STRING:
sortedSlice := t.values[index].([][]byte)
sortedSlice[i], sortedSlice[j] = sortedSlice[j], sortedSlice[i]
}
}
if t.bitMaps != nil {
for _, bitMap := range t.bitMaps {
if bitMap != nil {
isNilI := bitMap.IsMarked(i)
isNilJ := bitMap.IsMarked(j)
if isNilI {
bitMap.Mark(j)
} else {
bitMap.UnMark(j)
}
if isNilJ {
bitMap.Mark(i)
} else {
bitMap.UnMark(i)
}
}
}
}
t.timestamps[i], t.timestamps[j] = t.timestamps[j], t.timestamps[i]
}
func (t *Tablet) Less(i, j int) bool {
return t.timestamps[i] < t.timestamps[j]
}
func (t *Tablet) SetTimestamp(timestamp int64, rowIndex int) {
t.timestamps[rowIndex] = timestamp
}
// SetTimestampAt is a bounds-checked alternative to SetTimestamp. It validates
// rowIndex and returns an "illegal argument rowIndex" error instead of
// panicking with an index-out-of-range when rowIndex is negative or beyond the
// tablet's capacity, consistent with SetValueAt and GetValueAt. The existing
// SetTimestamp is preserved unchanged for source compatibility.
func (t *Tablet) SetTimestampAt(timestamp int64, rowIndex int) error {
if rowIndex < 0 || rowIndex >= t.maxRowNumber {
return fmt.Errorf("illegal argument rowIndex %d", rowIndex)
}
t.timestamps[rowIndex] = timestamp
return nil
}
func (t *Tablet) SetValueAt(value interface{}, 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 value == nil {
// Init the bitMap to mark nil value
if t.bitMaps == nil {
t.bitMaps = make([]*BitMap, len(t.values))
}
if t.bitMaps[columnIndex] == nil {
t.bitMaps[columnIndex] = NewBitMap(t.maxRowNumber)
}
// Mark the nil value position
t.bitMaps[columnIndex].Mark(rowIndex)
return nil
}
switch t.measurementSchemas[columnIndex].DataType {
case BOOLEAN:
values := t.values[columnIndex].([]bool)
switch v := value.(type) {
case bool:
values[rowIndex] = v
case *bool:
values[rowIndex] = *v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case INT32:
values := t.values[columnIndex].([]int32)
switch v := value.(type) {
case int32:
values[rowIndex] = v
case *int32:
values[rowIndex] = *v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case INT64, TIMESTAMP:
values := t.values[columnIndex].([]int64)
switch v := value.(type) {
case int64:
values[rowIndex] = v
case *int64:
values[rowIndex] = *v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case FLOAT:
values := t.values[columnIndex].([]float32)
switch v := value.(type) {
case float32:
values[rowIndex] = v
case *float32:
values[rowIndex] = *v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case DOUBLE:
values := t.values[columnIndex].([]float64)
switch v := value.(type) {
case float64:
values[rowIndex] = v
case *float64:
values[rowIndex] = *v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case TEXT, STRING:
values := t.values[columnIndex].([][]byte)
switch v := value.(type) {
case string:
values[rowIndex] = []byte(v)
case []byte:
values[rowIndex] = v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case BLOB:
values := t.values[columnIndex].([][]byte)
switch v := value.(type) {
case []byte:
values[rowIndex] = v
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
case DATE:
values := t.values[columnIndex].([]int32)
switch v := value.(type) {
case time.Time:
val, err := DateToInt32(v)
if err != nil {
return err
}
values[rowIndex] = val
default:
return fmt.Errorf("illegal argument value %v %v", value, reflect.TypeOf(value))
}
}
return nil
}
func (t *Tablet) GetMaxRowNumber() int {
return t.maxRowNumber
}
func (t *Tablet) GetValueAt(columnIndex, rowIndex int) (interface{}, error) {
if columnIndex < 0 || columnIndex >= len(t.measurementSchemas) {
return nil, fmt.Errorf("illegal argument columnIndex %d", columnIndex)
}
if rowIndex < 0 || rowIndex >= t.maxRowNumber {
return nil, fmt.Errorf("illegal argument rowIndex %d", rowIndex)
}
schema := t.measurementSchemas[columnIndex]
if t.bitMaps != nil && t.bitMaps[columnIndex] != nil && t.bitMaps[columnIndex].IsMarked(rowIndex) {
return nil, nil
}
switch schema.DataType {
case BOOLEAN:
return t.values[columnIndex].([]bool)[rowIndex], nil
case INT32:
return t.values[columnIndex].([]int32)[rowIndex], nil
case INT64, TIMESTAMP:
return t.values[columnIndex].([]int64)[rowIndex], nil
case FLOAT:
return t.values[columnIndex].([]float32)[rowIndex], nil
case DOUBLE:
return t.values[columnIndex].([]float64)[rowIndex], nil
case TEXT, STRING:
return string(t.values[columnIndex].([][]byte)[rowIndex]), nil
case BLOB:
return t.values[columnIndex].([][]byte)[rowIndex], nil
case DATE:
return Int32ToDate(t.values[columnIndex].([]int32)[rowIndex])
default:
return nil, fmt.Errorf("illegal datatype %v", schema.DataType)
}
}
func (t *Tablet) GetTimestampBytes() []byte {
buff := &bytes.Buffer{}
binary.Write(buff, binary.BigEndian, t.timestamps[0:t.RowSize])
return buff.Bytes()
}
func (t *Tablet) GetMeasurements() []string {
measurements := make([]string, len(t.measurementSchemas))
for i, s := range t.measurementSchemas {
measurements[i] = s.Measurement
}
return measurements
}
func (t *Tablet) getDataTypes() []int32 {
types := make([]int32, len(t.measurementSchemas))
for i, s := range t.measurementSchemas {
types[i] = int32(s.DataType)
}
return types
}
func (t *Tablet) getColumnCategories() []int8 {
columnCategories := make([]int8, len(t.columnCategories))
for i := range t.columnCategories {
columnCategories[i] = int8(t.columnCategories[i])
}
return columnCategories
}
func (t *Tablet) getValuesBytes() ([]byte, error) {
buff := &bytes.Buffer{}
for i, schema := range t.measurementSchemas {
switch schema.DataType {
case BOOLEAN:
binary.Write(buff, binary.BigEndian, t.values[i].([]bool)[0:t.RowSize])
case INT32, DATE:
binary.Write(buff, binary.BigEndian, t.values[i].([]int32)[0:t.RowSize])
case INT64, TIMESTAMP:
binary.Write(buff, binary.BigEndian, t.values[i].([]int64)[0:t.RowSize])
case FLOAT:
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:
for _, s := range t.values[i].([][]byte)[0:t.RowSize] {
binary.Write(buff, binary.BigEndian, int32(len(s)))
binary.Write(buff, binary.BigEndian, s)
}
default:
return nil, fmt.Errorf("illegal datatype %v", schema.DataType)
}
}
if t.bitMaps != nil {
for _, bitMap := range t.bitMaps {
columnHasNil := bitMap != nil && !bitMap.IsAllUnmarked()
binary.Write(buff, binary.BigEndian, columnHasNil)
if columnHasNil {
// Need to maintain consistency with the calculation method on the IoTDB side.
binary.Write(buff, binary.BigEndian, bitMap.GetBits()[0:t.RowSize/8+1])
}
}
}
return buff.Bytes(), nil
}
func (t *Tablet) Sort() error {
sort.Sort(t)
return nil
}
func (t *Tablet) Reset() {
t.RowSize = 0
t.bitMaps = nil
}
func NewTablet(insertTargetName string, measurementSchemas []*MeasurementSchema, maxRowNumber int) (*Tablet, error) {
tablet := &Tablet{
insertTargetName: insertTargetName,
measurementSchemas: measurementSchemas,
maxRowNumber: maxRowNumber,
}
tablet.timestamps = make([]int64, maxRowNumber)
tablet.values = make([]interface{}, len(measurementSchemas))
for i, schema := range tablet.measurementSchemas {
switch schema.DataType {
case BOOLEAN:
tablet.values[i] = make([]bool, maxRowNumber)
case INT32, DATE:
tablet.values[i] = make([]int32, maxRowNumber)
case INT64, TIMESTAMP:
tablet.values[i] = make([]int64, maxRowNumber)
case FLOAT:
tablet.values[i] = make([]float32, maxRowNumber)
case DOUBLE:
tablet.values[i] = make([]float64, maxRowNumber)
case TEXT, STRING, BLOB:
tablet.values[i] = make([][]byte, maxRowNumber)
default:
return nil, fmt.Errorf("illegal datatype %v", schema.DataType)
}
}
return tablet, nil
}
func NewRelationalTablet(tableName string, measurementSchemas []*MeasurementSchema, columnCategories []ColumnCategory, maxRowNumber int) (*Tablet, error) {
tablet, err := NewTablet(tableName, measurementSchemas, maxRowNumber)
if err != nil {
return nil, err
}
tablet.columnCategories = columnCategories
return tablet, nil
}
agents/go-common/iotdb-client-go/client/tls.go
+120
View File
@@ -0,0 +1,120 @@
/*
* 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 (
"crypto/tls"
"crypto/x509"
"fmt"
"net"
"os"
"time"
"github.com/apache/thrift/lib/go/thrift"
)
// TLSConfig enables TLS for an IoTDB client connection. Set CertFile and
// KeyFile together to enable mTLS client authentication.
type TLSConfig struct {
// Config is an optional base tls.Config. It is cloned before use.
Config *tls.Config
// CAFile is an optional PEM encoded CA certificate file used to verify the server.
CAFile string
// CertFile and KeyFile are optional PEM encoded client certificate and key files for mTLS.
CertFile string
KeyFile string
}
func newTransport(host string, port string, connectionTimeoutInMs int, tlsConfig *TLSConfig) (thrift.TTransport, error) {
conf := &thrift.TConfiguration{
ConnectTimeout: time.Duration(connectionTimeoutInMs) * time.Millisecond,
MaxFrameSize: thrift.DEFAULT_MAX_FRAME_SIZE,
}
hostPort := net.JoinHostPort(host, port)
var base thrift.TTransport
if tlsConfig == nil {
base = thrift.NewTSocketConf(hostPort, conf)
} else {
cfg, err := buildTLSConfig(tlsConfig)
if err != nil {
return nil, err
}
conf.TLSConfig = cfg
base = thrift.NewTSSLSocketConf(hostPort, conf)
}
return thrift.NewTFramedTransportConf(base, conf), nil
}
func buildTLSConfig(config *TLSConfig) (*tls.Config, error) {
if config == nil {
return nil, nil
}
tlsConfig := &tls.Config{}
if config.Config != nil {
tlsConfig = config.Config.Clone()
}
if config.CAFile != "" {
rootCAs, err := loadCertPool(tlsConfig.RootCAs, config.CAFile)
if err != nil {
return nil, err
}
tlsConfig.RootCAs = rootCAs
}
if config.CertFile != "" || config.KeyFile != "" {
if config.CertFile == "" || config.KeyFile == "" {
return nil, fmt.Errorf("both TLS CertFile and KeyFile must be set")
}
certificate, err := tls.LoadX509KeyPair(config.CertFile, config.KeyFile)
if err != nil {
return nil, fmt.Errorf("load TLS client certificate/key: %w", err)
}
tlsConfig.Certificates = append(tlsConfig.Certificates, certificate)
}
return tlsConfig, nil
}
func loadCertPool(base *x509.CertPool, caFile string) (*x509.CertPool, error) {
rootCAs := base
if rootCAs != nil {
rootCAs = rootCAs.Clone()
} else {
systemPool, err := x509.SystemCertPool()
if err == nil && systemPool != nil {
rootCAs = systemPool
} else {
rootCAs = x509.NewCertPool()
}
}
caCert, err := os.ReadFile(caFile)
if err != nil {
return nil, fmt.Errorf("read TLS CA file %q: %w", caFile, err)
}
if !rootCAs.AppendCertsFromPEM(caCert) {
return nil, fmt.Errorf("append TLS CA file %q: no certificates found", caFile)
}
return rootCAs, nil
}
agents/go-common/iotdb-client-go/client/tsblock.go
+162
View File
@@ -0,0 +1,162 @@
/*
* 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 (
"bytes"
"encoding/binary"
"fmt"
)
type TsBlock struct {
timeColumn Column
valueColumns []Column
positionCount int32
}
func NewTsBlock(positionCount int32, timeColumn Column, valueColumns ...Column) (*TsBlock, error) {
if valueColumns == nil {
return nil, fmt.Errorf("blocks is null")
}
return &TsBlock{
timeColumn: timeColumn,
valueColumns: valueColumns,
positionCount: positionCount,
}, nil
}
func DeserializeTsBlock(data []byte) (*TsBlock, error) {
// Serialized tsblock:
// +-------------+---------------+---------+------------+-----------+----------+
// | val col cnt | val col types | pos cnt | encodings | time col | val col |
// +-------------+---------------+---------+------------+-----------+----------+
// | int32 | list[byte] | int32 | list[byte] | bytes | bytes |
// +-------------+---------------+---------+------------+-----------+----------+
reader := bytes.NewReader(data)
// value column count
var valueColumnCount int32
if err := binary.Read(reader, binary.BigEndian, &valueColumnCount); err != nil {
return nil, err
}
// value column data types
valueColumnDataTypes := make([]TSDataType, valueColumnCount)
for i := int32(0); i < valueColumnCount; i++ {
dataType, err := deserializeDataType(reader)
if err != nil {
return nil, err
}
valueColumnDataTypes[i] = dataType
}
// position count
var positionCount int32
if err := binary.Read(reader, binary.BigEndian, &positionCount); err != nil {
return nil, err
}
// column encodings
columnEncodings := make([]ColumnEncoding, valueColumnCount+1)
for i := int32(0); i < valueColumnCount+1; i++ {
columnEncoding, err := deserializeColumnEncoding(reader)
if err != nil {
return nil, err
}
columnEncodings[i] = columnEncoding
}
// time column
timeColumnDecoder, err := getColumnDecoder(columnEncodings[0])
if err != nil {
return nil, err
}
timeColumn, err := timeColumnDecoder.ReadColumn(reader, INT64, positionCount)
if err != nil {
return nil, err
}
// value columns
valueColumns := make([]Column, valueColumnCount)
for i := int32(0); i < valueColumnCount; i++ {
valueColumnDecoder, err := getColumnDecoder(columnEncodings[i+1])
if err != nil {
return nil, err
}
valueColumn, err := valueColumnDecoder.ReadColumn(reader, valueColumnDataTypes[i], positionCount)
if err != nil {
return nil, err
}
valueColumns[i] = valueColumn
}
return NewTsBlock(positionCount, timeColumn, valueColumns...)
}
func deserializeDataType(reader *bytes.Reader) (TSDataType, error) {
b, err := reader.ReadByte()
if err != nil {
return UNKNOWN, err
}
return getDataTypeByByte(b)
}
func deserializeColumnEncoding(reader *bytes.Reader) (ColumnEncoding, error) {
b, err := reader.ReadByte()
if err != nil {
return RLE_COLUMN_ENCODING, err
}
return getColumnEncodingByByte(b)
}
func (t *TsBlock) GetPositionCount() int32 {
return t.positionCount
}
func (t *TsBlock) GetStartTime() (int64, error) {
return t.timeColumn.GetLong(0)
}
func (t *TsBlock) GetEndTime() (int64, error) {
return t.timeColumn.GetLong(t.positionCount - 1)
}
func (t *TsBlock) IsEmpty() bool {
return t.positionCount == 0
}
func (t *TsBlock) GetTimeByIndex(index int32) (int64, error) {
return t.timeColumn.GetLong(index)
}
func (t *TsBlock) GetValueColumnCount() int32 {
return int32(len(t.valueColumns))
}
func (t *TsBlock) GetTimeColumn() Column {
return t.timeColumn
}
func (t *TsBlock) GetValueColumns() *[]Column {
return &t.valueColumns
}
func (t *TsBlock) GetColumn(columnIndex int32) Column {
return t.valueColumns[columnIndex]
}
agents/go-common/iotdb-client-go/client/utils.go
+292
View File
@@ -0,0 +1,292 @@
/*
* 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 (
"bytes"
"encoding/binary"
"errors"
"fmt"
"strconv"
"time"
"github.com/apache/iotdb-client-go/v2/common"
"github.com/apache/iotdb-client-go/v2/rpc"
)
const (
DEFAULT_TIME_FORMAT = "default"
TIME_PRECISION = "timestamp_precision"
MILLISECOND = "ms"
MICROSECOND = "us"
NANOSECOND = "ns"
)
func getTimeFactor(openResp *rpc.TSOpenSessionResp) (int32, error) {
if !openResp.IsSetConfiguration() {
return 1_000, nil
}
precision, exists := openResp.GetConfiguration()[TIME_PRECISION]
if !exists {
return 1_000, nil
}
switch precision {
case MILLISECOND:
return 1_000, nil
case MICROSECOND:
return 1_000_000, nil
case NANOSECOND:
return 1_000_000_000, nil
default:
return 0, fmt.Errorf("unknown time precision: %v", precision)
}
}
func getTimePrecision(timeFactor int32) (string, error) {
switch timeFactor {
case 1_000:
return MILLISECOND, nil
case 1_000_000:
return MICROSECOND, nil
case 1_000_000_000:
return NANOSECOND, nil
default:
return "", fmt.Errorf("unknown time factor: %v", timeFactor)
}
}
func formatDatetime(timeFormat, timePrecision string, timestamp int64, zone *time.Location) string {
switch timeFormat {
case "long", "number":
return strconv.FormatInt(timestamp, 10)
case "default", "iso8601":
return parseLongToDateWithPrecision(timestamp, zone, timePrecision)
default:
sec := timestamp / 1000
nsec := (timestamp % 1000) * int64(time.Millisecond)
t := time.Unix(sec, nsec).In(zone)
return t.Format(timeFormat)
}
}
func getMilliSecond(timeValue int64, timeFactor int32) int64 {
return (timeValue / int64(timeFactor)) * 1_000
}
func getNanoSecond(timeValue int64, timeFactor int32) int {
return int((timeValue % int64(timeFactor)) * (1_000_000_000 / int64(timeFactor)))
}
func convertToTimestamp(timeValue int64, timeFactor int32) time.Time {
millis := getMilliSecond(timeValue, timeFactor)
nanos := getNanoSecond(timeValue, timeFactor)
return time.Unix(millis/1e3, (millis%1e3)*1e6).Add(time.Duration(nanos) * time.Nanosecond)
}
func parseLongToDateWithPrecision(timestamp int64, zone *time.Location, precision string) string {
var divisor int64
var digits int
switch precision {
case MILLISECOND:
divisor = 1000
digits = 3
case MICROSECOND:
divisor = 1_000_000
digits = 6
case NANOSECOND:
divisor = 1_000_000_000
digits = 9
default:
return ""
}
quotient := timestamp / divisor
remainder := timestamp % divisor
if timestamp < 0 && remainder != 0 {
quotient--
remainder += divisor
}
t := time.Unix(quotient, 0).In(zone)
year, month, day := t.Date()
hour, min, sec := t.Clock()
_, offset := t.Zone()
offsetSeconds := offset
sign := "+"
if offsetSeconds < 0 {
sign = "-"
offsetSeconds = -offsetSeconds
}
hours := offsetSeconds / 3600
minutes := (offsetSeconds % 3600) / 60
zoneOffset := fmt.Sprintf("%s%02d:%02d", sign, hours, minutes)
if offset == 0 {
zoneOffset = "Z"
}
formatStr := fmt.Sprintf("%%0%dd", digits)
fraction := fmt.Sprintf(formatStr, remainder)
isoStr := fmt.Sprintf("%04d-%02d-%02dT%02d:%02d:%02d.%s%s",
year, month, day, hour, min, sec, fraction, zoneOffset)
return isoStr
}
func int32ToString(n int32) string {
return strconv.Itoa(int(n))
}
func int64ToString(n int64) string {
return strconv.FormatInt(n, 10)
}
func float32ToString(val float32) string {
return strconv.FormatFloat(float64(val), 'f', -1, 32)
}
func float64ToString(val float64) string {
return strconv.FormatFloat(val, 'f', -1, 64)
}
func int32ToBytes(n int32) []byte {
bytesBuffer := bytes.NewBuffer([]byte{})
binary.Write(bytesBuffer, binary.BigEndian, n)
return bytesBuffer.Bytes()
}
func int64ToBytes(n int64) []byte {
bytesBuffer := bytes.NewBuffer([]byte{})
binary.Write(bytesBuffer, binary.BigEndian, n)
return bytesBuffer.Bytes()
}
func bytesToInt32(bys []byte) int32 {
bytesBuffer := bytes.NewBuffer(bys)
var data int32
binary.Read(bytesBuffer, binary.BigEndian, &data)
return data
}
func bytesToInt64(bys []byte) int64 {
bytesBuffer := bytes.NewBuffer(bys)
var data int64
binary.Read(bytesBuffer, binary.BigEndian, &data)
return data
}
func bytesToHexString(input []byte) string {
hexString := "0x"
if input != nil {
for _, b := range input {
hexString += fmt.Sprintf("%02x", b)
}
}
return hexString
}
func DateToInt32(localDate time.Time) (int32, error) {
if localDate.IsZero() {
return 0, errors.New("date expression is null or empty")
}
year := localDate.Year()
if year < 1000 || year > 9999 {
return 0, errors.New("year must be between 1000 and 9999")
}
// Convert to YYYY/MM/DD format
result := year*10000 + int(localDate.Month())*100 + localDate.Day()
return int32(result), nil
}
func Int32ToDate(val int32) (time.Time, error) {
date := int(val)
year := date / 10000
month := (date / 100) % 100
day := date % 100
localDate := time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.UTC)
if localDate.Year() != year || int(localDate.Month()) != month || localDate.Day() != day {
return time.Time{}, errors.New("invalid date format")
}
return localDate, nil
}
func bytesToDate(bys []byte) (time.Time, error) {
return Int32ToDate(bytesToInt32(bys))
}
func verifySuccesses(statuses []*common.TSStatus) error {
buff := bytes.Buffer{}
for _, status := range statuses {
if status.Code != SuccessStatus && status.Code != RedirectionRecommend {
buff.WriteString(*status.Message + ";")
}
}
errMsg := buff.String()
if len(errMsg) > 0 {
return &BatchError{statuses, errMsg}
}
return nil
}
func VerifySuccess(status *common.TSStatus) error {
if status.Code == RedirectionRecommend {
return nil
}
if status.Code == MultipleError {
if err := verifySuccesses(status.GetSubStatus()); err != nil {
return err
}
return nil
}
if status.Code != SuccessStatus {
msg := ""
if status.Message != nil {
msg = *status.Message
}
return &ExecutionError{Code: status.Code, Message: msg}
}
return nil
}
type Binary struct {
values []byte
}
func NewBinary(v []byte) *Binary {
return &Binary{v}
}
func (b *Binary) GetStringValue() string {
return string(b.values)
}
func (b *Binary) GetValues() []byte {
return b.values
}
agents/go-common/iotdb-client-go/common/GoUnusedProtection__.go
+6
View File
@@ -0,0 +1,6 @@
// Code generated by Thrift Compiler (0.23.0). DO NOT EDIT.
package common
var GoUnusedProtection__ int;
agents/go-common/iotdb-client-go/common/common-consts.go
+34
View File
@@ -0,0 +1,34 @@
// Code generated by Thrift Compiler (0.23.0). DO NOT EDIT.
package common
import (
"bytes"
"context"
"errors"
"fmt"
"iter"
"log/slog"
"time"
thrift "github.com/apache/thrift/lib/go/thrift"
"strings"
"regexp"
)
// (needed to ensure safety because of naive import list construction.)
var _ = bytes.Equal
var _ = context.Background
var _ = errors.New
var _ = fmt.Printf
var _ = iter.Pull[int]
var _ = slog.Log
var _ = time.Now
var _ = thrift.ZERO
// (needed by validator.)
var _ = strings.Contains
var _ = regexp.MatchString
func init() {
}
agents/go-common/iotdb-client-go/common/common.go
+9790
View File
File diff suppressed because it is too large Load Diff
agents/go-common/iotdb-client-go/go.mod
+5
View File
@@ -0,0 +1,5 @@
module github.com/apache/iotdb-client-go/v2
go 1.25
require github.com/apache/thrift v0.24.0
agents/go-common/iotdb-client-go/go.sum
+2
View File
@@ -0,0 +1,2 @@
github.com/apache/thrift v0.24.0 h1:zy31L1a49QTNB2bG1BBfMXol3yJrTH975G3pPubQVLQ=
github.com/apache/thrift v0.24.0/go.mod h1:zPt6WxgvTOM6hF92y8C+MkEM5LMxZuk4JcQOiU4Esvs=
agents/go-common/iotdb-client-go/rpc/GoUnusedProtection__.go
+6
View File
@@ -0,0 +1,6 @@
// Code generated by Thrift Compiler (0.23.0). DO NOT EDIT.
package rpc
var GoUnusedProtection__ int;
agents/go-common/iotdb-client-go/rpc/client-consts.go
+37
View File
@@ -0,0 +1,37 @@
// Code generated by Thrift Compiler (0.23.0). DO NOT EDIT.
package rpc
import (
"bytes"
"context"
"errors"
"fmt"
"iter"
"log/slog"
"time"
thrift "github.com/apache/thrift/lib/go/thrift"
"strings"
"regexp"
"github.com/apache/iotdb-client-go/v2/common"
)
// (needed to ensure safety because of naive import list construction.)
var _ = bytes.Equal
var _ = context.Background
var _ = errors.New
var _ = fmt.Printf
var _ = iter.Pull[int]
var _ = slog.Log
var _ = time.Now
var _ = thrift.ZERO
// (needed by validator.)
var _ = strings.Contains
var _ = regexp.MatchString
var _ = common.GoUnusedProtection__
func init() {
}
agents/go-common/iotdb-client-go/rpc/client.go
+48613
View File
File diff suppressed because it is too large Load Diff