You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2025/03/20 10:16:33 UTC
(iotdb) branch query_v2_py updated (945b5a43efc -> 1caad90573a)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch query_v2_py
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 945b5a43efc Try CI
new f1532d9e1b8 dev more
new 1caad90573a dev more
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client-py/iotdb/tsfile/utils/tsblock_serde.py | 21 ++---
.../client-py/iotdb/utils/IoTDBRpcDataSet.py | 97 ++++++++++------------
2 files changed, 49 insertions(+), 69 deletions(-)
(iotdb) 01/02: dev more
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch query_v2_py
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f1532d9e1b85c07efd446547b48008b65c64f9df
Author: HTHou <ha...@apache.org>
AuthorDate: Thu Mar 20 16:20:16 2025 +0800
dev more
---
.../client-py/iotdb/utils/IoTDBRpcDataSet.py | 47 ++++++++++++++--------
1 file changed, 30 insertions(+), 17 deletions(-)
diff --git a/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py b/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
index 2d30d30c66d..c4e36a0f794 100644
--- a/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
+++ b/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
@@ -22,6 +22,7 @@ import logging
import numpy as np
import pandas as pd
+from datetime import date
from thrift.transport import TTransport
from iotdb.thrift.rpc.IClientRPCService import TSFetchResultsReq, TSCloseOperationReq
from iotdb.tsfile.utils.date_utils import parse_int_to_date
@@ -103,6 +104,7 @@ class IoTDBRpcDataSet(object):
self.__query_result = query_result
if query_result is not None:
self.__query_result_size = len(query_result)
+ self.__query_result_index = 0
self.__is_closed = False
self.__empty_resultSet = False
self.__rows_index = 0
@@ -148,16 +150,15 @@ class IoTDBRpcDataSet(object):
if (
self.has_cached_data_frame
or self.__query_result is None
- or len(self.__query_result) == 0
+ or len(self.__query_result) <= self.__query_result_index
):
return
- binary_size = len(self.__query_result)
- binary_index = 0
result = {}
time_column_values, column_values, null_indicators, _ = deserialize(
- self.__query_result[0]
+ self.__query_result[self.__query_result_index]
)
- self.__query_result = None
+ self.__query_result[self.__query_result_index] = None
+ self.__query_result_index += 1
time_array = np.frombuffer(
time_column_values, np.dtype(np.longlong).newbyteorder(">")
)
@@ -191,7 +192,7 @@ class IoTDBRpcDataSet(object):
)
# BOOLEAN
elif data_type == 0:
- data_array = np.frombuffer(value_buffer, np.dtype("?"))
+ data_array = np.array(value_buffer).astype("bool")
# INT32, DATE
elif data_type == 1 or data_type == 9:
data_array = np.frombuffer(
@@ -228,11 +229,16 @@ class IoTDBRpcDataSet(object):
null_indicator = null_indicators[location]
- if len(data_array) < total_length:
+ if len(data_array) < total_length or (
+ data_type == 0 and null_indicator is not None
+ ):
tmp_array = np.full(total_length, None, dtype=object)
if null_indicator is not None:
indexes = [not v for v in null_indicator]
- tmp_array[indexes] = data_array
+ if data_type == 0:
+ tmp_array[indexes] = data_array[indexes]
+ else:
+ tmp_array[indexes] = data_array
# INT32, DATE
if data_type == 1 or data_type == 9:
@@ -247,7 +253,6 @@ class IoTDBRpcDataSet(object):
result[i + 1] = data_array
self.data_frame = pd.DataFrame(result, dtype=object)
- self.__query_result = None
if not self.data_frame.empty:
self.has_cached_data_frame = True
@@ -255,7 +260,9 @@ class IoTDBRpcDataSet(object):
return self.has_cached_data_frame
def _has_next_result_set(self):
- if (self.__query_result is not None) and (len(self.__query_result[0]) != 0):
+ if (self.__query_result is not None) and (
+ len(self.__query_result) > self.__query_result_index
+ ):
return True
if self.__empty_resultSet:
return False
@@ -269,9 +276,10 @@ class IoTDBRpcDataSet(object):
result[column_name] = []
while self._has_next_result_set():
time_column_values, column_values, null_indicators, _ = deserialize(
- self.__query_result[0]
+ self.__query_result[self.__query_result_index]
)
- self.__query_result = None
+ self.__query_result[self.__query_result_index] = None
+ self.__query_result_index += 1
time_array = np.frombuffer(
time_column_values, np.dtype(np.longlong).newbyteorder(">")
)
@@ -310,7 +318,7 @@ class IoTDBRpcDataSet(object):
)
# BOOLEAN
elif data_type == 0:
- data_array = np.frombuffer(value_buffer, np.dtype("?"))
+ data_array = np.array(value_buffer).astype("bool")
# INT32
elif data_type == 1:
data_array = np.frombuffer(
@@ -364,13 +372,15 @@ class IoTDBRpcDataSet(object):
data_array = pd.Series(data_array).apply(parse_int_to_date)
else:
raise RuntimeError("unsupported data type {}.".format(data_type))
- if data_array.dtype.byteorder == ">":
+ if data_array.dtype.byteorder == ">" and len(data_array) > 0:
data_array = data_array.byteswap().view(
data_array.dtype.newbyteorder("<")
)
tmp_array = []
null_indicator = null_indicators[location]
- if len(data_array) < total_length:
+ if len(data_array) < total_length or (
+ data_type == 0 and null_indicator is not None
+ ):
# BOOLEAN, INT32, INT64, TIMESTAMP
if (
data_type == 0
@@ -391,11 +401,14 @@ class IoTDBRpcDataSet(object):
or data_type == 10
or data_type == 9
):
- tmp_array = np.full(total_length, None, dtype=data_array.dtype)
+ tmp_array = np.full(total_length, None, dtype=object)
if null_indicator is not None:
indexes = [not v for v in null_indicator]
- tmp_array[indexes] = data_array
+ if data_type == 0:
+ tmp_array[indexes] = data_array[indexes]
+ else:
+ tmp_array[indexes] = data_array
if data_type == 1:
tmp_array = pd.Series(tmp_array).astype("Int32")
(iotdb) 02/02: dev more
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch query_v2_py
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1caad90573a0fed4772760fb48e941f5661448ce
Author: HTHou <ha...@apache.org>
AuthorDate: Thu Mar 20 18:16:14 2025 +0800
dev more
---
.../client-py/iotdb/tsfile/utils/tsblock_serde.py | 21 +++------
.../client-py/iotdb/utils/IoTDBRpcDataSet.py | 50 ++++++----------------
2 files changed, 19 insertions(+), 52 deletions(-)
diff --git a/iotdb-client/client-py/iotdb/tsfile/utils/tsblock_serde.py b/iotdb-client/client-py/iotdb/tsfile/utils/tsblock_serde.py
index be86c26666c..ccc091d791b 100644
--- a/iotdb-client/client-py/iotdb/tsfile/utils/tsblock_serde.py
+++ b/iotdb-client/client-py/iotdb/tsfile/utils/tsblock_serde.py
@@ -271,7 +271,9 @@ def deserialize(buffer):
position_count, buffer = read_int_from_buffer(buffer)
column_encodings, buffer = read_column_encoding(buffer, value_column_count + 1)
- time_column_values, buffer = read_time_column(buffer, position_count)
+ time_column_values, _, buffer = read_column(
+ column_encodings[0], buffer, TSDataType.INT64, position_count
+ )
column_values = [None] * value_column_count
null_indicators = [None] * value_column_count
for i in range(value_column_count):
@@ -384,15 +386,6 @@ def deserialize_null_indicators(buffer, size):
# +---------------+-----------------+-------------+
-def read_time_column(buffer, size):
- null_indicators, buffer = deserialize_null_indicators(buffer, size)
- if null_indicators is None:
- values, buffer = read_from_buffer(buffer, size * 8)
- else:
- raise Exception("TimeColumn should not contains null value")
- return values, buffer
-
-
def read_int64_column(buffer, data_type, position_count):
null_indicators, buffer = deserialize_null_indicators(buffer, position_count)
if null_indicators is None:
@@ -537,7 +530,7 @@ def read_run_length_column(buffer, data_type, position_count):
return (
repeat(column, data_type, position_count),
- null_indicators * position_count,
+ None if null_indicators is None else null_indicators * position_count,
buffer,
)
@@ -546,7 +539,7 @@ def repeat(buffer, data_type, position_count):
if data_type == TSDataType.BOOLEAN or data_type == TSDataType.TEXT:
return buffer * position_count
else:
- res = bytes(0)
+ res = bytearray()
for _ in range(position_count):
- res.join(buffer)
- return res
+ res.extend(buffer if isinstance(buffer, bytes) else bytes(buffer))
+ return bytes(res)
diff --git a/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py b/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
index c4e36a0f794..c9e45304707 100644
--- a/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
+++ b/iotdb-client/client-py/iotdb/utils/IoTDBRpcDataSet.py
@@ -205,20 +205,11 @@ class IoTDBRpcDataSet(object):
)
# TEXT, STRING, BLOB
elif data_type == 5 or data_type == 11 or data_type == 10:
- j = 0
- offset = 0
+ index = 0
data_array = []
- while offset < value_buffer_len:
- length = int.from_bytes(
- value_buffer[offset : offset + 4],
- byteorder="big",
- signed=False,
- )
- offset += 4
- value = bytes(value_buffer[offset : offset + length])
- data_array.append(value)
- j += 1
- offset += length
+ while index < value_buffer_len:
+ data_array.append(value_buffer[index])
+ index += 1
data_array = np.array(data_array, dtype=object)
else:
raise RuntimeError("unsupported data type {}.".format(data_type))
@@ -331,38 +322,21 @@ class IoTDBRpcDataSet(object):
)
# TEXT, STRING
elif data_type == 5 or data_type == 11:
- j = 0
- offset = 0
+ index = 0
data_array = []
- while offset < value_buffer_len:
- length = int.from_bytes(
- value_buffer[offset : offset + 4],
- byteorder="big",
- signed=False,
- )
- offset += 4
- value_bytes = bytes(value_buffer[offset : offset + length])
+ while index < value_buffer_len:
+ value_bytes = value_buffer[index]
value = value_bytes.decode("utf-8")
data_array.append(value)
- j += 1
- offset += length
+ index += 1
data_array = pd.Series(data_array).astype(str)
# BLOB
elif data_type == 10:
- j = 0
- offset = 0
+ index = 0
data_array = []
- while offset < value_buffer_len:
- length = int.from_bytes(
- value_buffer[offset : offset + 4],
- byteorder="big",
- signed=False,
- )
- offset += 4
- value = value_buffer[offset : offset + length]
- data_array.append(value)
- j += 1
- offset += length
+ while index < value_buffer_len:
+ data_array.append(value_buffer[index])
+ index += 1
data_array = pd.Series(data_array)
# DATE
elif data_type == 9: