diff --git a/documentation/webdocs/markdowndocs/Connector.md b/documentation/webdocs/markdowndocs/Connector.md index 79e98efdc63b91bc952f0f5f459f3e0603325494..a0433d1f09d7c5f2ec1205f89d2efe638703dc7d 100644 --- a/documentation/webdocs/markdowndocs/Connector.md +++ b/documentation/webdocs/markdowndocs/Connector.md @@ -175,26 +175,34 @@ TDengine provides APIs for continuous query driven by time, which run queries pe ### C/C++ subscription API -For the time being, TDengine supports subscription on one table. It is implemented through periodic pulling from a TDengine server. +For the time being, TDengine supports subscription on one or multiple tables. It is implemented through periodic pulling from a TDengine server. -- `TAOS_SUB *taos_subscribe(char *host, char *user, char *pass, char *db, char *table, int64_t time, int mseconds)` - The API is used to start a subscription session by given a handle. The parameters required are _host_ (IP address of a TDenginer server), _user_ (username), _pass_ (password), _db_ (database to use), _table_ (table name to subscribe), _time_ (start time to subscribe, 0 for now), _mseconds_ (pulling period). If failed to open a subscription session, a _NULL_ pointer is returned. +* `TAOS_SUB *taos_subscribe(TAOS* taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval)` + The API is used to start a subscription session, it returns the subscription object on success and `NULL` in case of failure, the parameters are: + * **taos**: The database connnection, which must be established already. + * **restart**: `Zero` to continue a subscription if it already exits, other value to start from the beginning. + * **topic**: The unique identifier of a subscription. + * **sql**: A sql statement for data query, it can only be a `select` statement, can only query for raw data, and can only query data in ascending order of the timestamp field. + * **fp**: A callback function to receive query result, only used in asynchronization mode and should be `NULL` in synchronization mode, please refer below for its prototype. + * **param**: User provided additional parameter for the callback function. + * **interval**: Pulling interval in millisecond. Under asynchronization mode, API will call the callback function `fp` in this interval, system performance will be impacted if this interval is too short. Under synchronization mode, if the duration between two call to `taos_consume` is less than this interval, the second call blocks until the duration exceed this interval. -- `TAOS_ROW taos_consume(TAOS_SUB *tsub)` - The API used to get the new data from a TDengine server. It should be put in an infinite loop. The parameter _tsub_ is the handle returned by _taos_subscribe_. If new data are updated, the API will return a row of the result. Otherwise, the API is blocked until new data arrives. If _NULL_ pointer is returned, it means an error occurs. +* `typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code)` + Prototype of the callback function, the parameters are: + * tsub: The subscription object. + * res: The query result. + * param: User provided additional parameter when calling `taos_subscribe`. + * code: Error code in case of failures. -- `void taos_unsubscribe(TAOS_SUB *tsub)` - Stop a subscription session by the handle returned by _taos_subscribe_. - - -- `int taos_num_subfields(TAOS_SUB *tsub)` - The API used to get the number of fields in a row. +* `TAOS_RES *taos_consume(TAOS_SUB *tsub)` + The API used to get the new data from a TDengine server. It should be put in an loop. The parameter `tsub` is the handle returned by `taos_subscribe`. This API should only be called in synchronization mode. If the duration between two call to `taos_consume` is less than pulling interval, the second call blocks until the duration exceed the interval. The API returns the new rows if new data arrives, or empty rowset otherwise, and if there's an error, it returns `NULL`. + +* `void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress)` -- `TAOS_FIELD *taos_fetch_subfields(TAOS_SUB *tsub)` - The API used to get the description of each column. + Stop a subscription session by the handle returned by `taos_subscribe`. If `keepProgress` is **not** zero, the subscription progress information is kept and can be reused in later call to `taos_subscribe`, the information is removed otherwise. ## Java Connector @@ -590,6 +598,28 @@ c1.execute('select * from tb') for data in c1: print("ts=%s, temperature=%d, humidity=%f" %(data[0], data[1],data[2]) ``` + +* create a subscription +```python +# Create a subscription with topic 'test' and consumption interval 1000ms. +# The first argument is True means to restart the subscription; +# if the subscription with topic 'test' has already been created, then pass +# False to this argument means to continue the existing subscription. +sub = conn.subscribe(True, "test", "select * from meters;", 1000) +``` + +* consume a subscription +```python +data = sub.consume() +for d in data: + print(d) +``` + +* close the subscription +```python +sub.close() +``` + * close the connection ```python c1.close() diff --git a/documentation/webdocs/markdowndocs/administrator-ch.md b/documentation/webdocs/markdowndocs/administrator-ch.md index 17acbacfcf89119c5d4f5cad02428e977d27db2e..35beb610f2789ea9d521eaa0314160a4ddff4025 100644 --- a/documentation/webdocs/markdowndocs/administrator-ch.md +++ b/documentation/webdocs/markdowndocs/administrator-ch.md @@ -337,7 +337,8 @@ TDengine也支持在shell对已存在的表从CSV文件中进行数据导入。 insert into tb1 file a.csv b.csv tb2 c.csv … import into tb1 file a.csv b.csv tb2 c.csv … ``` -> 注意：导入的CSV文件不能够带表头, 且表的列与CSV文件的列需要严格对应 +> 注意：导入的CSV文件不能够带表头, 且表的列与CSV文件的列需要严格对应。 +> 同样还可以使用[样例数据导入工具][1]对数据进行横向和纵向扩展导入。 ## 数据导出 @@ -408,3 +409,6 @@ KILL STREAM TDengine启动后，会自动创建一个监测数据库`LOG`，并自动将服务器的CPU、内存、硬盘空间、带宽、请求数、磁盘读写速度、慢查询等信息定时写入该数据库。TDengine还将重要的系统操作（比如登录、创建、删除数据库等）日志以及各种错误报警信息记录下来存放在`LOG`库里。系统管理员可以通过客户端程序查看记录库中的运行负载信息，（在企业版中）还可以通过浏览器查看数据的图标可视化结果。 这些监测信息的采集缺省是打开的，但可以修改配置文件里的选项`monitor`将其关闭或打开。 + + +[1]: https://github.com/taosdata/TDengine/tree/develop/importSampleData \ No newline at end of file diff --git a/documentation/webdocs/markdowndocs/advanced features-ch.md b/documentation/webdocs/markdowndocs/advanced features-ch.md index 14a2801209c9b92ed16d38ed220cee5c3684cd4d..4d01eaf364cfe17b4cb4658dc8596fbb52a65ae2 100644 --- a/documentation/webdocs/markdowndocs/advanced features-ch.md +++ b/documentation/webdocs/markdowndocs/advanced features-ch.md @@ -63,28 +63,11 @@ CREATE TABLE QUERY_RES ## 数据订阅(Publisher/Subscriber) 基于数据天然的时间序列特性，TDengine的数据写入（insert）与消息系统的数据发布（pub）逻辑上一致，均可视为系统中插入一条带时间戳的新记录。同时，TDengine在内部严格按照数据时间序列单调递增的方式保存数据。本质上来说，TDengine中里每一张表均可视为一个标准的消息队列。 -TDengine内嵌支持轻量级的消息订阅与推送服务。使用系统提供的API，用户可订阅数据库中的某一张表（或超级表）。订阅的逻辑和操作状态的维护均是由客户端完成，客户端定时轮询服务器是否有新的记录到达，有新的记录到达就会将结果反馈到客户。 +TDengine内嵌支持轻量级的消息订阅与推送服务。使用系统提供的API，用户可使用普通查询语句订阅数据库中的一张或多张表。订阅的逻辑和操作状态的维护均是由客户端完成，客户端定时轮询服务器是否有新的记录到达，有新的记录到达就会将结果反馈到客户。 TDengine的订阅与推送服务的状态是客户端维持，TDengine服务器并不维持。因此如果应用重启，从哪个时间点开始获取最新数据，由应用决定。 -#### API说明 - -使用订阅的功能，主要API如下： - - -示例代码：请看安装包中的的示范程序 +订阅相关API请见 [连接器](https://www.taosdata.com/cn/documentation/connector/)。 ## 缓存 (Cache) TDengine采用时间驱动缓存管理策略（First-In-First-Out，FIFO），又称为写驱动的缓存管理机制。这种策略有别于读驱动的数据缓存模式（Least-Recent-Use，LRU），直接将最近写入的数据保存在系统的缓存中。当缓存达到临界值的时候，将最早的数据批量写入磁盘。一般意义上来说，对于物联网数据的使用，用户最为关心最近产生的数据，即当前状态。TDengine充分利用了这一特性，将最近到达的（当前状态）数据保存在缓存中。 diff --git a/documentation/webdocs/markdowndocs/connector-ch.md b/documentation/webdocs/markdowndocs/connector-ch.md index 9b4dc5a2e5a04a3d158d6cd2f9d04e407523b145..b5d8fb5afb12ede82f2cdcd9ea29e20e8a82d6b8 100644 --- a/documentation/webdocs/markdowndocs/connector-ch.md +++ b/documentation/webdocs/markdowndocs/connector-ch.md @@ -164,27 +164,36 @@ TDengine提供时间驱动的实时流式计算API。可以每隔一指定的时 ### C/C++ 数据订阅接口 -订阅API目前支持订阅一张表，并通过定期轮询的方式不断获取写入表中的最新数据。 +订阅API目前支持订阅一张或多张表，并通过定期轮询的方式不断获取写入表中的最新数据。 -- `TAOS_SUB *taos_subscribe(char *host, char *user, char *pass, char *db, char *table, int64_t time, int mseconds)` +* `TAOS_SUB *taos_subscribe(TAOS* taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval)` - 该API用来启动订阅，需要提供的参数包含：TDengine管理主节点的IP地址、用户名、密码、数据库、数据库表的名字；time是开始订阅消息的时间，是从1970年1月1日起计算的毫秒数，为长整型, 如果设为0，表示从当前时间开始订阅；mseconds为查询数据库更新的时间间隔，单位为毫秒，建议设为1000毫秒。返回值为一指向TDengine_SUB结构的指针，如果返回为空，表示失败。 + 该函数负责启动订阅服务，成功时返回订阅对象，失败时返回 `NULL`，其参数为： + * taos：已经建立好的数据库连接 + * restart：如果订阅已经存在，是重新开始，还是继续之前的订阅 + * topic：订阅的主题（即名称），此参数是订阅的唯一标识 + * sql：订阅的查询语句，此语句只能是 `select` 语句，只应查询原始数据，只能按时间正序查询数据 + * fp：收到查询结果时的回调函数（稍后介绍函数原型），只在异步调用时使用，同步调用时此参数应该传 `NULL` + * param：调用回调函数时的附加参数，系统API将其原样传递到回调函数，不进行任何处理 + * interval：轮询周期，单位为毫秒。异步调用时，将根据此参数周期性的调用回调函数，为避免对系统性能造成影响，不建议将此参数设置的过小；同步调用时，如两次调用`taos_consume`的间隔小于此周期，API将会阻塞，直到时间间隔超过此周期。 -- `TAOS_ROW taos_consume(TAOS_SUB *tsub)` +* `typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code)` - 该API用来获取最新消息，应用程序一般会将其置于一个无限循环语句中。其中参数tsub是taos_subscribe的返回值。如果数据库有新的记录，该API将返回，返回参数是一行记录。如果没有新的记录，该API将阻塞。如果返回值为空，说明系统出错，需要检查系统是否还在正常运行。 + 异步模式下，回调函数的原型，其参数为： + * tsub：订阅对象 + * res：查询结果集，注意结果集中可能没有记录 + * param：调用 `taos_subscribe`时客户程序提供的附加参数 + * code：错误码 -- `void taos_unsubscribe(TAOS_SUB *tsub)` - 该API用于取消订阅，参数tsub是taos_subscribe的返回值。应用程序退出时，需要调用该API，否则有资源泄露。 +* `TAOS_RES *taos_consume(TAOS_SUB *tsub)` -- `int taos_num_subfields(TAOS_SUB *tsub)` + 同步模式下，该函数用来获取订阅的结果。 用户应用程序将其置于一个循环之中。 如两次调用`taos_consume`的间隔小于订阅的轮询周期，API将会阻塞，直到时间间隔超过此周期。 如果数据库有新记录到达，该API将返回该最新的记录，否则返回一个没有记录的空结果集。 如果返回值为 `NULL`，说明系统出错。 异步模式下，用户程序不应调用此API。 - 该API用来获取返回的一排数据中数据的列数 +* `void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress)` -- `TAOS_FIELD *taos_fetch_subfields(TAOS_SUB *tsub)` + 取消订阅。 如参数 `keepProgress` 不为0，API会保留订阅的进度信息，后续调用 `taos_subscribe` 时可以基于此进度继续；否则将删除进度信息，后续只能重新开始读取数据。 - 该API用来获取每列数据的属性（数据类型、名字、字节数），与taos_num_subfields配合使用，可用来解析返回的一排数据。 ## Java Connector @@ -205,7 +214,7 @@ TDengine 的 JDBC 驱动实现尽可能的与关系型数据库驱动保持一 * TDengine 不提供针对单条数据记录的删除和修改的操作，驱动中也没有支持相关方法。 * 由于不支持删除和修改，所以也不支持事务操作。 * 目前不支持表间的 union 操作。 -* 目前不支持嵌套查询(nested query)，`对每个 Connection 的实例，至多只能有一个打开的 ResultSet 实例；如果在 ResultSet还没关闭的情况下执行了新的查询，TSDBJDBCDriver 则会自动关闭上一个 ResultSet`。 +* 目前不支持嵌套查询(nested query)，对每个 Connection 的实例，至多只能有一个打开的 ResultSet 实例；如果在 ResultSet还没关闭的情况下执行了新的查询，TSDBJDBCDriver 则会自动关闭上一个 ResultSet。 ## TAOS-JDBCDriver 版本以及支持的 TDengine 版本和 JDK 版本 @@ -586,6 +595,27 @@ c1.execute('select * from tb') for data in c1: print("ts=%s, temperature=%d, humidity=%f" %(data[0], data[1],data[2]) ``` + +* 创建订阅 +```python +# 创建一个主题为 'test' 消费周期为1000毫秒的订阅 +# 第一个参数为 True 表示重新开始订阅，如为 False 且之前创建过主题为 'test' 的订阅，则表示继续消费此订阅的数据，而不是重新开始消费所有数据 +sub = conn.subscribe(True, "test", "select * from meters;", 1000) +``` + +* 消费订阅的数据 +```python +data = sub.consume() +for d in data: + print(d) +``` + +* 取消订阅 +```python +sub.close() +``` + + * 关闭连接 ```python c1.close() diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index bc26abe0a432814f6499e943aeaa2ce6b4f42476..fcbd3dac1f27a2bfadc470fcbab7075c1fab6246 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -278,6 +278,7 @@ typedef struct { int rspType; int rspLen; uint64_t qhandle; + int64_t uid; int64_t useconds; int64_t offset; // offset value from vnode during projection query of stable int row; @@ -324,6 +325,7 @@ typedef struct _sql_obj { uint32_t queryId; void * thandle; void * pStream; + void * pSubscription; char * sqlstr; char retry; char maxRetry; @@ -410,6 +412,12 @@ void tscDestroyResPointerInfo(SSqlRes *pRes); void tscFreeSqlCmdData(SSqlCmd *pCmd); void tscFreeResData(SSqlObj* pSql); +/** + * free query result of the sql object + * @param pObj + */ +void tscFreeSqlResult(SSqlObj* pSql); + /** * only free part of resources allocated during query. * Note: this function is multi-thread safe. diff --git a/src/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h b/src/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h index 958252b4deca7708c99e6b762613813c2f9d330b..8dbe63d75a73dd18a15bc1da8f99c7b8db774eea 100644 --- a/src/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h +++ b/src/client/jni/com_taosdata_jdbc_TSDBJNIConnector.h @@ -135,7 +135,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_closeConnectionIm * Signature: (Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;JI)J */ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp - (JNIEnv *, jobject, jstring, jstring, jstring, jstring, jstring, jlong, jint); + (JNIEnv *, jobject, jlong, jboolean, jstring, jstring, jint); /* * Class: com_taosdata_jdbc_TSDBJNIConnector @@ -143,7 +143,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp * Signature: (J)Lcom/taosdata/jdbc/TSDBResultSetRowData; */ JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp - (JNIEnv *, jobject, jlong); + (JNIEnv *, jobject, jlong, jint); /* * Class: com_taosdata_jdbc_TSDBJNIConnector @@ -151,7 +151,7 @@ JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp * Signature: (J)V */ JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_unsubscribeImp - (JNIEnv *, jobject, jlong); + (JNIEnv *, jobject, jlong, jboolean); /* * Class: com_taosdata_jdbc_TSDBJNIConnector diff --git a/src/client/src/TSDBJNIConnector.c b/src/client/src/TSDBJNIConnector.c index 9cec4f4b0f2e4f67448f35afddaec14e92bce7cf..228403c79d318d922f5571a9663b3c97bbffbbc8 100644 --- a/src/client/src/TSDBJNIConnector.c +++ b/src/client/src/TSDBJNIConnector.c @@ -20,6 +20,7 @@ #include "tscJoinProcess.h" #include "tsclient.h" #include "tscUtil.h" +#include "ttime.h" int __init = 0; @@ -514,92 +515,42 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_closeConnectionIm } } -JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp(JNIEnv *env, jobject jobj, jstring jhost, - jstring juser, jstring jpass, jstring jdb, - jstring jtable, jlong jtime, - jint jperiod) { - TAOS_SUB *tsub; - jlong sub = 0; - char * host = NULL; - char * user = NULL; - char * pass = NULL; - char * db = NULL; - char * table = NULL; - int64_t time = 0; - int period = 0; +JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp(JNIEnv *env, jobject jobj, jlong con, + jboolean restart, jstring jtopic, jstring jsql, jint jinterval) { + jlong sub = 0; + TAOS *taos = (TAOS *)con; + char *topic = NULL; + char *sql = NULL; jniGetGlobalMethod(env); jniTrace("jobj:%p, in TSDBJNIConnector_subscribeImp", jobj); - if (jhost != NULL) { - host = (char *)(*env)->GetStringUTFChars(env, jhost, NULL); - } - if (juser != NULL) { - user = (char *)(*env)->GetStringUTFChars(env, juser, NULL); - } - if (jpass != NULL) { - pass = (char *)(*env)->GetStringUTFChars(env, jpass, NULL); - } - if (jdb != NULL) { - db = (char *)(*env)->GetStringUTFChars(env, jdb, NULL); - } - if (jtable != NULL) { - table = (char *)(*env)->GetStringUTFChars(env, jtable, NULL); + if (jtopic != NULL) { + topic = (char *)(*env)->GetStringUTFChars(env, jtopic, NULL); } - time = (int64_t)jtime; - period = (int)jperiod; - - if (user == NULL) { - jniTrace("jobj:%p, user is null, use tsDefaultUser", jobj); - user = tsDefaultUser; - } - if (pass == NULL) { - jniTrace("jobj:%p, pass is null, use tsDefaultPass", jobj); - pass = tsDefaultPass; + if (jsql != NULL) { + sql = (char *)(*env)->GetStringUTFChars(env, jsql, NULL); } - jniTrace("jobj:%p, host:%s, user:%s, pass:%s, db:%s, table:%s, time:%d, period:%d", jobj, host, user, pass, db, table, - time, period); - tsub = taos_subscribe(host, user, pass, db, table, time, period); + TAOS_SUB *tsub = taos_subscribe(taos, (int)restart, topic, sql, NULL, NULL, jinterval); sub = (jlong)tsub; if (sub == 0) { - jniTrace("jobj:%p, failed to subscribe to db:%s, table:%s", jobj, db, table); + jniTrace("jobj:%p, failed to subscribe: topic:%s", jobj, jtopic); } else { - jniTrace("jobj:%p, successfully subscribe to db:%s, table:%s, sub:%ld, tsub:%p", jobj, db, table, sub, tsub); + jniTrace("jobj:%p, successfully subscribe: topic: %s", jobj, jtopic); } - if (host != NULL) (*env)->ReleaseStringUTFChars(env, jhost, host); - if (user != NULL && user != tsDefaultUser) (*env)->ReleaseStringUTFChars(env, juser, user); - if (pass != NULL && pass != tsDefaultPass) (*env)->ReleaseStringUTFChars(env, jpass, pass); - if (db != NULL) (*env)->ReleaseStringUTFChars(env, jdb, db); - if (table != NULL) (*env)->ReleaseStringUTFChars(env, jtable, table); + if (topic != NULL) (*env)->ReleaseStringUTFChars(env, jtopic, topic); + if (sql != NULL) (*env)->ReleaseStringUTFChars(env, jsql, sql); return sub; } -JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNIEnv *env, jobject jobj, jlong sub) { - jniTrace("jobj:%p, in TSDBJNIConnector_consumeImp, sub:%ld", jobj, sub); - - TAOS_SUB * tsub = (TAOS_SUB *)sub; - TAOS_ROW row = taos_consume(tsub); - TAOS_FIELD *fields = taos_fetch_subfields(tsub); - int num_fields = taos_subfields_count(tsub); - - jniGetGlobalMethod(env); - - jniTrace("jobj:%p, check fields:%p, num_fields=%d", jobj, fields, num_fields); - +static jobject convert_one_row(JNIEnv *env, TAOS_ROW row, TAOS_FIELD* fields, int num_fields) { jobject rowobj = (*env)->NewObject(env, g_rowdataClass, g_rowdataConstructor, num_fields); jniTrace("created a rowdata object, rowobj:%p", rowobj); - if (row == NULL) { - jniTrace("jobj:%p, tsub:%p, fields size is %d, fetch row to the end", jobj, tsub, num_fields); - return NULL; - } - - char tmp[TSDB_MAX_BYTES_PER_ROW] = {0}; - for (int i = 0; i < num_fields; i++) { if (row[i] == NULL) { continue; @@ -634,6 +585,7 @@ JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNI } break; case TSDB_DATA_TYPE_BINARY: { + char tmp[TSDB_MAX_BYTES_PER_ROW] = {0}; strncpy(tmp, row[i], (size_t) fields[i].bytes); // handle the case that terminated does not exist (*env)->CallVoidMethod(env, rowobj, g_rowdataSetStringFp, i, (*env)->NewStringUTF(env, tmp)); @@ -642,7 +594,7 @@ JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNI } case TSDB_DATA_TYPE_NCHAR: (*env)->CallVoidMethod(env, rowobj, g_rowdataSetByteArrayFp, i, - jniFromNCharToByteArray(env, (char*)row[i], fields[i].bytes)); + jniFromNCharToByteArray(env, (char*)row[i], fields[i].bytes)); break; case TSDB_DATA_TYPE_TIMESTAMP: (*env)->CallVoidMethod(env, rowobj, g_rowdataSetTimestampFp, i, (jlong) * ((int64_t *)row[i])); @@ -651,13 +603,56 @@ JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNI break; } } - jniTrace("jobj:%p, rowdata retrieved, rowobj:%p", jobj, rowobj); return rowobj; } -JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_unsubscribeImp(JNIEnv *env, jobject jobj, jlong sub) { +JNIEXPORT jobject JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNIEnv *env, jobject jobj, jlong sub, jint timeout) { + jniTrace("jobj:%p, in TSDBJNIConnector_consumeImp, sub:%ld", jobj, sub); + jniGetGlobalMethod(env); + + TAOS_SUB *tsub = (TAOS_SUB *)sub; + jobject rows = (*env)->NewObject(env, g_arrayListClass, g_arrayListConstructFp); + + int64_t start = taosGetTimestampMs(); + int count = 0; + + while (true) { + TAOS_RES * res = taos_consume(tsub); + if (res == NULL) { + jniError("jobj:%p, tsub:%p, taos_consume returns NULL", jobj, tsub); + return NULL; + } + + TAOS_FIELD *fields = taos_fetch_fields(res); + int num_fields = taos_num_fields(res); + while (true) { + TAOS_ROW row = taos_fetch_row(res); + if (row == NULL) { + break; + } + jobject rowobj = convert_one_row(env, row, fields, num_fields); + (*env)->CallBooleanMethod(env, rows, g_arrayListAddFp, rowobj); + count++; + } + + if (count > 0) { + break; + } + if (timeout == -1) { + continue; + } + if (((int)(taosGetTimestampMs() - start)) >= timeout) { + jniTrace("jobj:%p, sub:%ld, timeout", jobj, sub); + break; + } + } + + return rows; +} + +JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_unsubscribeImp(JNIEnv *env, jobject jobj, jlong sub, jboolean keepProgress) { TAOS_SUB *tsub = (TAOS_SUB *)sub; - taos_unsubscribe(tsub); + taos_unsubscribe(tsub, keepProgress); } JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_validateCreateTableSqlImp(JNIEnv *env, jobject jobj, diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 01a0da4e0c7080c7f86cb0b0b233d9291b726cfa..a25979031ab4ae24a5ead49a17d96dd74c3aa0bf 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -41,7 +41,10 @@ int (*tscProcessMsgRsp[TSDB_SQL_MAX])(SSqlObj *pSql); char *doBuildMsgHeader(SSqlObj *pSql, char **pStart); void (*tscUpdateVnodeMsg[TSDB_SQL_MAX])(SSqlObj *pSql, char *buf); void tscProcessActivityTimer(void *handle, void *tmrId); -int tscKeepConn[TSDB_SQL_MAX] = {0}; +int tscKeepConn[TSDB_SQL_MAX] = {0}; +TSKEY tscGetSubscriptionProgress(void* sub, int64_t uid); +void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts); +void tscSaveSubscriptionProgress(void* sub); static int32_t minMsgSize() { return tsRpcHeadSize + sizeof(STaosDigest); } @@ -399,7 +402,7 @@ void *tscProcessMsgFromServer(char *msg, void *ahandle, void *thandle) { SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0); if (msg == NULL) { - tscTrace("%p no response from ip:0x%x", pSql, pSql->ip); + tscTrace("%p no response from ip:%s", pSql, taosIpStr(pSql->ip)); pSql->index++; pSql->thandle = NULL; @@ -1571,7 +1574,7 @@ static char *doSerializeTableInfo(SSqlObj *pSql, int32_t numOfMeters, int32_t vn SMeterSidExtInfo *pMeterInfo = (SMeterSidExtInfo *)pMsg; pMeterInfo->sid = htonl(pMeterMeta->sid); pMeterInfo->uid = htobe64(pMeterMeta->uid); - + pMeterInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pMeterMeta->uid)); pMsg += sizeof(SMeterSidExtInfo); } else { SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pMeterMetaInfo->vnodeIndex); @@ -1582,7 +1585,8 @@ static char *doSerializeTableInfo(SSqlObj *pSql, int32_t numOfMeters, int32_t vn pMeterInfo->sid = htonl(pQueryMeterInfo->sid); pMeterInfo->uid = htobe64(pQueryMeterInfo->uid); - + pMeterInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pQueryMeterInfo->uid)); + pMsg += sizeof(SMeterSidExtInfo); memcpy(pMsg, pQueryMeterInfo->tags, pMetricMeta->tagLen); @@ -3415,12 +3419,28 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) { pRes->numOfRows = htonl(pRetrieve->numOfRows); pRes->precision = htons(pRetrieve->precision); pRes->offset = htobe64(pRetrieve->offset); - pRes->useconds = htobe64(pRetrieve->useconds); pRes->data = pRetrieve->data; - + SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); tscSetResultPointer(pQueryInfo, pRes); + + if (pSql->pSubscription != NULL) { + TAOS_FIELD *pField = tscFieldInfoGetField(pCmd, pCmd->fieldsInfo.numOfOutputCols - 1); + int16_t offset = tscFieldInfoGetOffset(pCmd, pCmd->fieldsInfo.numOfOutputCols - 1); + char* p = pRes->data + (pField->bytes + offset) * pRes->numOfRows; + + int32_t numOfMeters = htonl(*(int32_t*)p); + p += sizeof(int32_t); + for (int i = 0; i < numOfMeters; i++) { + int64_t uid = htobe64(*(int64_t*)p); + p += sizeof(int64_t); + TSKEY key = htobe64(*(TSKEY*)p); + p += sizeof(TSKEY); + tscUpdateSubscriptionProgress(pSql->pSubscription, uid, key); + } + } + pRes->row = 0; /** diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 93e09f082812ef7e91ec66a34cc37513b4098ece..2b72c24c94ba986e71cc8bca73d630a18e9785a8 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -13,8 +13,8 @@ * along with this program. If not, see . */ -#include "hash.h" #include "os.h" +#include "hash.h" #include "tcache.h" #include "tlog.h" #include "tnote.h" @@ -64,8 +64,15 @@ TAOS *taos_connect_imp(const char *ip, const char *user, const char *pass, const } if (ip && ip[0]) { + tscMgmtIpList.numOfIps = 4; strcpy(tscMgmtIpList.ipstr[0], ip); tscMgmtIpList.ip[0] = inet_addr(ip); + strcpy(tscMgmtIpList.ipstr[1], ip); + tscMgmtIpList.ip[1] = inet_addr(ip); + strcpy(tscMgmtIpList.ipstr[2], tsMasterIp); + tscMgmtIpList.ip[2] = inet_addr(tsMasterIp); + strcpy(tscMgmtIpList.ipstr[3], tsSecondIp); + tscMgmtIpList.ip[3] = inet_addr(tsSecondIp); } pObj = (STscObj *)malloc(sizeof(STscObj)); @@ -694,7 +701,7 @@ int taos_select_db(TAOS *taos, const char *db) { return taos_query(taos, sql); } -void taos_free_result(TAOS_RES *res) { +void taos_free_result_imp(TAOS_RES* res, int keepCmd) { if (res == NULL) return; SSqlObj *pSql = (SSqlObj *)res; @@ -712,6 +719,8 @@ void taos_free_result(TAOS_RES *res) { pSql->thandle = NULL; tscFreeSqlObj(pSql); tscTrace("%p Async SqlObj is freed by app", pSql); + } else if (keepCmd) { + tscFreeSqlResult(pSql); } else { tscFreeSqlObjPartial(pSql); } @@ -767,8 +776,13 @@ void taos_free_result(TAOS_RES *res) { * Then this object will be reused and no free operation is required. */ pSql->thandle = NULL; - tscFreeSqlObjPartial(pSql); - tscTrace("%p sql result is freed by app", pSql); + if (keepCmd) { + tscFreeSqlResult(pSql); + tscTrace("%p sql result is freed by app while sql command is kept", pSql); + } else { + tscFreeSqlObjPartial(pSql); + tscTrace("%p sql result is freed by app", pSql); + } } } else { // if no free resource msg is sent to vnode, we free this object immediately. @@ -778,6 +792,9 @@ void taos_free_result(TAOS_RES *res) { assert(pRes->numOfRows == 0 || (pCmd->command > TSDB_SQL_LOCAL)); tscFreeSqlObj(pSql); tscTrace("%p Async sql result is freed by app", pSql); + } else if (keepCmd) { + tscFreeSqlResult(pSql); + tscTrace("%p sql result is freed while sql command is kept", pSql); } else { tscFreeSqlObjPartial(pSql); tscTrace("%p sql result is freed", pSql); @@ -785,6 +802,10 @@ void taos_free_result(TAOS_RES *res) { } } +void taos_free_result(TAOS_RES *res) { + taos_free_result_imp(res, 0); +} + int taos_errno(TAOS *taos) { STscObj *pObj = (STscObj *)taos; int code; @@ -891,59 +912,61 @@ void taos_stop_query(TAOS_RES *res) { int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) { int len = 0; for (int i = 0; i < num_fields; ++i) { + if (i > 0) { + str[len++] = ' '; + } + if (row[i] == NULL) { - len += sprintf(str + len, "%s ", TSDB_DATA_NULL_STR); + len += sprintf(str + len, "%s", TSDB_DATA_NULL_STR); continue; } switch (fields[i].type) { case TSDB_DATA_TYPE_TINYINT: - len += sprintf(str + len, "%d ", *((char *)row[i])); + len += sprintf(str + len, "%d", *((char *)row[i])); break; case TSDB_DATA_TYPE_SMALLINT: - len += sprintf(str + len, "%d ", *((short *)row[i])); + len += sprintf(str + len, "%d", *((short *)row[i])); break; case TSDB_DATA_TYPE_INT: - len += sprintf(str + len, "%d ", *((int *)row[i])); + len += sprintf(str + len, "%d", *((int *)row[i])); break; case TSDB_DATA_TYPE_BIGINT: - len += sprintf(str + len, "%" PRId64 " ", *((int64_t *)row[i])); + len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i])); break; case TSDB_DATA_TYPE_FLOAT: { float fv = 0; fv = GET_FLOAT_VAL(row[i]); - len += sprintf(str + len, "%f ", fv); + len += sprintf(str + len, "%f", fv); } break; case TSDB_DATA_TYPE_DOUBLE: { double dv = 0; dv = GET_DOUBLE_VAL(row[i]); - len += sprintf(str + len, "%lf ", dv); + len += sprintf(str + len, "%lf", dv); } break; case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_NCHAR: { - /* limit the max length of string to no greater than the maximum length, - * in case of not null-terminated string */ - size_t xlen = strlen(row[i]); - size_t trueLen = MIN(xlen, fields[i].bytes); - - memcpy(str + len, (char *)row[i], trueLen); - - str[len + trueLen] = ' '; - len += (trueLen + 1); - } break; + size_t xlen = 0; + for (xlen = 0; xlen <= fields[i].bytes; xlen++) { + char c = ((char*)row[i])[xlen]; + if (c == 0) break; + str[len++] = c; + } + str[len] = 0; + } break; case TSDB_DATA_TYPE_TIMESTAMP: - len += sprintf(str + len, "%" PRId64 " ", *((int64_t *)row[i])); + len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i])); break; case TSDB_DATA_TYPE_BOOL: - len += sprintf(str + len, "%d ", *((int8_t *)row[i])); + len += sprintf(str + len, "%d", *((int8_t *)row[i])); default: break; } diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index f2e9395c68b0dfc5a057b331cf00d38dbd9cb311..8a2c8ae44e0281b3d2f91b5c4a0b4151b7db0e37 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -22,125 +22,407 @@ #include "tsclient.h" #include "tsocket.h" #include "ttime.h" +#include "ttimer.h" #include "tutil.h" +#include "tscUtil.h" +#include "tcache.h" -typedef struct { - void * signature; - char name[TSDB_METER_ID_LEN]; - int mseconds; - TSKEY lastKey; - uint64_t stime; - TAOS_FIELD fields[TSDB_MAX_COLUMNS]; - int numOfFields; - TAOS * taos; - TAOS_RES * result; +typedef struct SSubscriptionProgress { + int64_t uid; + TSKEY key; +} SSubscriptionProgress; + +typedef struct SSub { + void * signature; + char topic[32]; + int64_t lastSyncTime; + int64_t lastConsumeTime; + TAOS * taos; + void * pTimer; + SSqlObj * pSql; + int interval; + TAOS_SUBSCRIBE_CALLBACK fp; + void * param; + int numOfMeters; + SSubscriptionProgress * progress; } SSub; -TAOS_SUB *taos_subscribe(const char *host, const char *user, const char *pass, const char *db, const char *name, int64_t time, int mseconds) { - SSub *pSub; - pSub = (SSub *)malloc(sizeof(SSub)); - if (pSub == NULL) return NULL; - memset(pSub, 0, sizeof(SSub)); +static int tscCompareSubscriptionProgress(const void* a, const void* b) { + const SSubscriptionProgress* x = (const SSubscriptionProgress*)a; + const SSubscriptionProgress* y = (const SSubscriptionProgress*)b; + if (x->uid > y->uid) return 1; + if (x->uid < y->uid) return -1; + return 0; +} + +TSKEY tscGetSubscriptionProgress(void* sub, int64_t uid) { + if (sub == NULL) + return 0; + + SSub* pSub = (SSub*)sub; + for (int s = 0, e = pSub->numOfMeters; s < e;) { + int m = (s + e) / 2; + SSubscriptionProgress* p = pSub->progress + m; + if (p->uid > uid) + e = m; + else if (p->uid < uid) + s = m + 1; + else + return p->key; + } + + return 0; +} + +void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts) { + if( sub == NULL) + return; + + SSub* pSub = (SSub*)sub; + for (int s = 0, e = pSub->numOfMeters; s < e;) { + int m = (s + e) / 2; + SSubscriptionProgress* p = pSub->progress + m; + if (p->uid > uid) + e = m; + else if (p->uid < uid) + s = m + 1; + else { + if (ts >= p->key) p->key = ts; + break; + } + } +} + +static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* sql) { + SSub* pSub = calloc(1, sizeof(SSub)); + if (pSub == NULL) { + globalCode = TSDB_CODE_CLI_OUT_OF_MEMORY; + tscError("failed to allocate memory for subscription"); + return NULL; + } + + SSqlObj* pSql = calloc(1, sizeof(SSqlObj)); + if (pSql == NULL) { + globalCode = TSDB_CODE_CLI_OUT_OF_MEMORY; + tscError("failed to allocate SSqlObj for subscription"); + goto failed; + } + + pSql->signature = pSql; + pSql->pTscObj = pObj; + + char* sqlstr = (char*)malloc(strlen(sql) + 1); + if (sqlstr == NULL) { + tscError("failed to allocate sql string for subscription"); + goto failed; + } + strcpy(sqlstr, sql); + strtolower(sqlstr, sqlstr); + pSql->sqlstr = sqlstr; + + tsem_init(&pSql->rspSem, 0, 0); + tsem_init(&pSql->emptyRspSem, 0, 1); + + SSqlRes *pRes = &pSql->res; + pRes->numOfRows = 1; + pRes->numOfTotal = 0; + + pSql->pSubscription = pSub; + pSub->pSql = pSql; pSub->signature = pSub; - strcpy(pSub->name, name); - pSub->mseconds = mseconds; - pSub->lastKey = time; - if (pSub->lastKey == 0) { - pSub->lastKey = taosGetTimestampMs(); + strncpy(pSub->topic, topic, sizeof(pSub->topic)); + pSub->topic[sizeof(pSub->topic) - 1] = 0; + return pSub; + +failed: + if (sqlstr != NULL) { + free(sqlstr); + } + if (pSql != NULL) { + free(pSql); + } + free(pSub); + return NULL; +} + + +static void tscProcessSubscriptionTimer(void *handle, void *tmrId) { + SSub *pSub = (SSub *)handle; + if (pSub == NULL || pSub->pTimer != tmrId) return; + + TAOS_RES* res = taos_consume(pSub); + if (res != NULL) { + pSub->fp(pSub, res, pSub->param, 0); + } + + taosTmrReset(tscProcessSubscriptionTimer, pSub->interval, pSub, tscTmr, &pSub->pTimer); +} + + +int tscUpdateSubscription(STscObj* pObj, SSub* pSub) { + int code = (uint8_t)tsParseSql(pSub->pSql, pObj->acctId, pObj->db, false); + if (code != TSDB_CODE_SUCCESS) { + tscError("failed to parse sql statement: %s", pSub->topic); + return 0; + } + + SSqlCmd* pCmd = &pSub->pSql->cmd; + if (pCmd->command != TSDB_SQL_SELECT) { + tscError("only 'select' statement is allowed in subscription: %s", pSub->topic); + return 0; + } + + SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0); + int numOfMeters = 0; + if (!UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) { + SMetricMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta; + for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) { + SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i); + numOfMeters += pVnodeSidList->numOfSids; + } } - taos_init(); - pSub->taos = taos_connect(host, user, pass, NULL, 0); - if (pSub->taos == NULL) { - tfree(pSub); + SSubscriptionProgress* progress = (SSubscriptionProgress*)calloc(numOfMeters, sizeof(SSubscriptionProgress)); + if (progress == NULL) { + tscError("failed to allocate memory for progress: %s", pSub->topic); + return 0; + } + + if (UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) { + numOfMeters = 1; + int64_t uid = pMeterMetaInfo->pMeterMeta->uid; + progress[0].uid = uid; + progress[0].key = tscGetSubscriptionProgress(pSub, uid); } else { - char qstr[256] = {0}; - sprintf(qstr, "use %s", db); - int res = taos_query(pSub->taos, qstr); - if (res != 0) { - tscError("failed to open DB:%s", db); - taos_close(pSub->taos); - tfree(pSub); - } else { - snprintf(qstr, tListLen(qstr), "select * from %s where _c0 > now+1000d", pSub->name); - if (taos_query(pSub->taos, qstr)) { - tscTrace("failed to select, reason:%s", taos_errstr(pSub->taos)); - taos_close(pSub->taos); - tfree(pSub); - return NULL; + SMetricMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta; + numOfMeters = 0; + for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) { + SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i); + for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) { + SMeterSidExtInfo *pMeterInfo = tscGetMeterSidInfo(pVnodeSidList, j); + int64_t uid = pMeterInfo->uid; + progress[numOfMeters].uid = uid; + progress[numOfMeters++].key = tscGetSubscriptionProgress(pSub, uid); } - pSub->result = taos_use_result(pSub->taos); - pSub->numOfFields = taos_num_fields(pSub->result); - memcpy(pSub->fields, taos_fetch_fields(pSub->result), sizeof(TAOS_FIELD) * pSub->numOfFields); } + qsort(progress, numOfMeters, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress); } - return pSub; + free(pSub->progress); + pSub->numOfMeters = numOfMeters; + pSub->progress = progress; + + pSub->lastSyncTime = taosGetTimestampMs(); + + return 1; } -TAOS_ROW taos_consume(TAOS_SUB *tsub) { - SSub * pSub = (SSub *)tsub; - TAOS_ROW row; - char qstr[256]; - if (pSub == NULL) return NULL; - if (pSub->signature != pSub) return NULL; - - while (1) { - if (pSub->result != NULL) { - row = taos_fetch_row(pSub->result); - if (row != NULL) { - pSub->lastKey = *((uint64_t *)row[0]); - return row; - } +static int tscLoadSubscriptionProgress(SSub* pSub) { + char buf[TSDB_MAX_SQL_LEN]; + sprintf(buf, "%s/subscribe/%s", dataDir, pSub->topic); - taos_free_result(pSub->result); - pSub->result = NULL; - uint64_t etime = taosGetTimestampMs(); - int64_t mseconds = pSub->mseconds - etime + pSub->stime; - if (mseconds < 0) mseconds = 0; - taosMsleep((int)mseconds); - } + FILE* fp = fopen(buf, "r"); + if (fp == NULL) { + tscTrace("subscription progress file does not exist: %s", pSub->topic); + return 1; + } - pSub->stime = taosGetTimestampMs(); + if (fgets(buf, sizeof(buf), fp) == NULL) { + tscTrace("invalid subscription progress file: %s", pSub->topic); + fclose(fp); + return 0; + } - sprintf(qstr, "select * from %s where _c0 > %" PRId64 " order by _c0 asc", pSub->name, pSub->lastKey); - if (taos_query(pSub->taos, qstr)) { - tscTrace("failed to select, reason:%s", taos_errstr(pSub->taos)); - return NULL; + for (int i = 0; i < sizeof(buf); i++) { + if (buf[i] == 0) + break; + if (buf[i] == '\r' || buf[i] == '\n') { + buf[i] = 0; + break; } + } + if (strcmp(buf, pSub->pSql->sqlstr) != 0) { + tscTrace("subscription sql statement mismatch: %s", pSub->topic); + fclose(fp); + return 0; + } - pSub->result = taos_use_result(pSub->taos); + if (fgets(buf, sizeof(buf), fp) == NULL || atoi(buf) < 0) { + tscTrace("invalid subscription progress file: %s", pSub->topic); + fclose(fp); + return 0; + } - if (pSub->result == NULL) { - tscTrace("failed to get result, reason:%s", taos_errstr(pSub->taos)); - return NULL; + int numOfMeters = atoi(buf); + SSubscriptionProgress* progress = calloc(numOfMeters, sizeof(SSubscriptionProgress)); + for (int i = 0; i < numOfMeters; i++) { + if (fgets(buf, sizeof(buf), fp) == NULL) { + fclose(fp); + free(progress); + return 0; } + int64_t uid, key; + sscanf(buf, "%" SCNd64 ":%" SCNd64, &uid, &key); + progress[i].uid = uid; + progress[i].key = key; } - return NULL; + fclose(fp); + + qsort(progress, numOfMeters, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress); + pSub->numOfMeters = numOfMeters; + pSub->progress = progress; + tscTrace("subscription progress loaded, %d tables: %s", numOfMeters, pSub->topic); + return 1; } -void taos_unsubscribe(TAOS_SUB *tsub) { - SSub *pSub = (SSub *)tsub; +void tscSaveSubscriptionProgress(void* sub) { + SSub* pSub = (SSub*)sub; - if (pSub == NULL) return; - if (pSub->signature != pSub) return; + char path[256]; + sprintf(path, "%s/subscribe", dataDir); + if (access(path, 0) != 0) { + mkdir(path, 0777); + } - taos_close(pSub->taos); - free(pSub); + sprintf(path, "%s/subscribe/%s", dataDir, pSub->topic); + FILE* fp = fopen(path, "w+"); + if (fp == NULL) { + tscError("failed to create progress file for subscription: %s", pSub->topic); + return; + } + + fputs(pSub->pSql->sqlstr, fp); + fprintf(fp, "\n%d\n", pSub->numOfMeters); + for (int i = 0; i < pSub->numOfMeters; i++) { + int64_t uid = pSub->progress[i].uid; + TSKEY key = pSub->progress[i].key; + fprintf(fp, "%" PRId64 ":%" PRId64 "\n", uid, key); + } + + fclose(fp); } -int taos_subfields_count(TAOS_SUB *tsub) { +TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval) { + STscObj* pObj = (STscObj*)taos; + if (pObj == NULL || pObj->signature != pObj) { + globalCode = TSDB_CODE_DISCONNECTED; + tscError("connection disconnected"); + return NULL; + } + + SSub* pSub = tscCreateSubscription(pObj, topic, sql); + if (pSub == NULL) { + return NULL; + } + pSub->taos = taos; + + if (restart) { + tscTrace("restart subscription: %s", topic); + } else { + tscLoadSubscriptionProgress(pSub); + } + + if (!tscUpdateSubscription(pObj, pSub)) { + taos_unsubscribe(pSub, 1); + return NULL; + } + + pSub->interval = interval; + if (fp != NULL) { + tscTrace("asynchronize subscription, create new timer", topic); + pSub->fp = fp; + pSub->param = param; + taosTmrReset(tscProcessSubscriptionTimer, interval, pSub, tscTmr, &pSub->pTimer); + } + + return pSub; +} + +void taos_free_result_imp(SSqlObj* pSql, int keepCmd); + +TAOS_RES *taos_consume(TAOS_SUB *tsub) { SSub *pSub = (SSub *)tsub; + if (pSub == NULL) return NULL; - return pSub->numOfFields; + tscSaveSubscriptionProgress(pSub); + + SSqlObj* pSql = pSub->pSql; + SSqlRes *pRes = &pSql->res; + + if (pSub->pTimer == NULL) { + int64_t duration = taosGetTimestampMs() - pSub->lastConsumeTime; + if (duration < (int64_t)(pSub->interval)) { + tscTrace("subscription consume too frequently, blocking..."); + taosMsleep(pSub->interval - (int32_t)duration); + } + } + + for (int retry = 0; retry < 3; retry++) { + tscRemoveFromSqlList(pSql); + + if (taosGetTimestampMs() - pSub->lastSyncTime > 10 * 60 * 1000) { + tscTrace("begin meter synchronization"); + char* sqlstr = pSql->sqlstr; + pSql->sqlstr = NULL; + taos_free_result_imp(pSql, 0); + pSql->sqlstr = sqlstr; + taosClearDataCache(tscCacheHandle); + if (!tscUpdateSubscription(pSub->taos, pSub)) return NULL; + tscTrace("meter synchronization completed"); + } else { + uint16_t type = pSql->cmd.type; + taos_free_result_imp(pSql, 1); + pRes->numOfRows = 1; + pRes->numOfTotal = 0; + pRes->qhandle = 0; + pSql->thandle = NULL; + pSql->cmd.command = TSDB_SQL_SELECT; + pSql->cmd.type = type; + + tscGetMeterMetaInfo(&pSql->cmd, 0)->vnodeIndex = 0; + } + + tscDoQuery(pSql); + if (pRes->code != TSDB_CODE_NOT_ACTIVE_TABLE) { + break; + } + // meter was removed, make sync time zero, so that next retry will + // do synchronization first + pSub->lastSyncTime = 0; + } + + if (pRes->code != TSDB_CODE_SUCCESS) { + tscError("failed to query data, error code=%d", pRes->code); + tscRemoveFromSqlList(pSql); + return NULL; + } + + pSub->lastConsumeTime = taosGetTimestampMs(); + return pSql; } -TAOS_FIELD *taos_fetch_subfields(TAOS_SUB *tsub) { +void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) { SSub *pSub = (SSub *)tsub; + if (pSub == NULL || pSub->signature != pSub) return; - return pSub->fields; + if (pSub->pTimer != NULL) { + taosTmrStop(pSub->pTimer); + } + + if (keepProgress) { + tscSaveSubscriptionProgress(pSub); + } else { + char path[256]; + sprintf(path, "%s/subscribe/%s", dataDir, pSub->topic); + remove(path); + } + + tscFreeSqlObj(pSub->pSql); + free(pSub->progress); + memset(pSub, 0, sizeof(*pSub)); + free(pSub); } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 31a4433fb5d2082c81fadbc3f0812f31f369f03b..8c5953c10c4c429a720b229f1ecd069ea9e5c6ce 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -13,9 +13,9 @@ * along with this program. If not, see . */ +#include "os.h" #include "tscUtil.h" #include "hash.h" -#include "os.h" #include "taosmsg.h" #include "tcache.h" #include "tkey.h" @@ -425,6 +425,21 @@ void tscFreeResData(SSqlObj* pSql) { tscDestroyResPointerInfo(pRes); } +void tscFreeSqlResult(SSqlObj* pSql) { + tfree(pSql->res.pRsp); + pSql->res.row = 0; + pSql->res.numOfRows = 0; + pSql->res.numOfTotal = 0; + + pSql->res.numOfGroups = 0; + tfree(pSql->res.pGroupRec); + + tscDestroyLocalReducer(pSql); + + tscDestroyResPointerInfo(&pSql->res); + tfree(pSql->res.pColumnIndex); +} + void tscFreeSqlObjPartial(SSqlObj* pSql) { if (pSql == NULL || pSql->signature != pSql) { return; @@ -445,9 +460,8 @@ void tscFreeSqlObjPartial(SSqlObj* pSql) { pthread_mutex_lock(&pObj->mutex); tfree(pSql->sqlstr); pthread_mutex_unlock(&pObj->mutex); - - tscFreeResData(pSql); - + + tscFreeSqlResult(pSql); tfree(pSql->pSubs); pSql->numOfSubs = 0; @@ -855,7 +869,7 @@ void tscFieldInfoSetValFromField(SFieldInfo* pFieldInfo, int32_t index, TAOS_FIE } void tscFieldInfoUpdateVisible(SFieldInfo* pFieldInfo, int32_t index, bool visible) { - if (index < 0 || index > pFieldInfo->numOfOutputCols) { + if (index < 0 || index >= pFieldInfo->numOfOutputCols) { return; } diff --git a/src/connector/python/linux/python2/taos/cinterface.py b/src/connector/python/linux/python2/taos/cinterface.py index 86a3489d0789d19a73f5318563569d4527845313..505619436cc1ad5d01a4134aede29477c6f6ae48 100644 --- a/src/connector/python/linux/python2/taos/cinterface.py +++ b/src/connector/python/linux/python2/taos/cinterface.py @@ -13,14 +13,14 @@ def _convert_microsecond_to_datetime(micro): def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False): """Function to convert C bool row to python row """ - _timstamp_converter = _convert_millisecond_to_datetime + _timestamp_converter = _convert_millisecond_to_datetime if micro: - _timstamp_converter = _convert_microsecond_to_datetime + _timestamp_converter = _convert_microsecond_to_datetime if num_of_rows > 0: - return list(map(_timstamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)][::-1])) + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)][::-1])) else: - return list(map(_timstamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)])) + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[:abs(num_of_rows)])) def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False): """Function to convert C bool row to python row @@ -144,6 +144,8 @@ class CTaosInterface(object): libtaos.taos_use_result.restype = ctypes.c_void_p libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) libtaos.taos_errstr.restype = ctypes.c_char_p + libtaos.taos_subscribe.restype = ctypes.c_void_p + libtaos.taos_consume.restype = ctypes.c_void_p def __init__(self, config=None): ''' @@ -252,6 +254,41 @@ class CTaosInterface(object): """ return CTaosInterface.libtaos.taos_affected_rows(connection) + @staticmethod + def subscribe(connection, restart, topic, sql, interval): + """Create a subscription + @restart boolean, + @sql string, sql statement for data query, must be a 'select' statement. + @topic string, name of this subscription + """ + return ctypes.c_void_p(CTaosInterface.libtaos.taos_subscribe( + connection, + 1 if restart else 0, + ctypes.c_char_p(topic.encode('utf-8')), + ctypes.c_char_p(sql.encode('utf-8')), + None, + None, + interval)) + + @staticmethod + def consume(sub): + """Consume data of a subscription + """ + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_consume(sub)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.libtaos.taos_num_fields(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + return result, fields + + @staticmethod + def unsubscribe(sub, keepProgress): + """Cancel a subscription + """ + CTaosInterface.libtaos.taos_unsubscribe(sub, 1 if keepProgress else 0) + @staticmethod def useResult(connection): '''Use result after calling self.query @@ -275,8 +312,8 @@ class CTaosInterface(object): if num_of_rows == 0: return None, 0 - blocks = [None] * len(fields) isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) + blocks = [None] * len(fields) for i in range(len(fields)): data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] @@ -351,4 +388,20 @@ class CTaosInterface(object): def errStr(connection): """Return the error styring """ - return CTaosInterface.libtaos.taos_errstr(connection) \ No newline at end of file + return CTaosInterface.libtaos.taos_errstr(connection) + + +if __name__ == '__main__': + cinter = CTaosInterface() + conn = cinter.connect() + + print('Query return value: {}'.format(cinter.query(conn, 'show databases'))) + print('Affected rows: {}'.format(cinter.affectedRows(conn))) + + result, des = CTaosInterface.useResult(conn) + + data, num_of_rows = CTaosInterface.fetchBlock(result, des) + + print(data) + + cinter.close(conn) \ No newline at end of file diff --git a/src/connector/python/linux/python2/taos/connection.py b/src/connector/python/linux/python2/taos/connection.py index ba24209552600d6ee75258f929eeff829dd7b675..04fbbdec04144624a0b1f4ba25083a91ade21cce 100644 --- a/src/connector/python/linux/python2/taos/connection.py +++ b/src/connector/python/linux/python2/taos/connection.py @@ -1,5 +1,5 @@ -# from .cursor import TDengineCursor from .cursor import TDengineCursor +from .subscription import TDengineSubscription from .cinterface import CTaosInterface class TDengineConnection(object): @@ -50,6 +50,14 @@ class TDengineConnection(object): """ return CTaosInterface.close(self._conn) + def subscribe(self, restart, topic, sql, interval): + """Create a subscription. + """ + if self._conn is None: + return None + sub = CTaosInterface.subscribe(self._conn, restart, topic, sql, interval) + return TDengineSubscription(sub) + def cursor(self): """Return a new Cursor object using the connection. """ diff --git a/src/connector/python/linux/python2/taos/subscription.py b/src/connector/python/linux/python2/taos/subscription.py new file mode 100644 index 0000000000000000000000000000000000000000..2d01395532820c3bd0e068ef7eb3d425eaaa6d78 --- /dev/null +++ b/src/connector/python/linux/python2/taos/subscription.py @@ -0,0 +1,52 @@ +from .cinterface import CTaosInterface +from .error import * + +class TDengineSubscription(object): + """TDengine subscription object + """ + def __init__(self, sub): + self._sub = sub + + + def consume(self): + """Consume rows of a subscription + """ + if self._sub is None: + raise OperationalError("Invalid use of consume") + + result, fields = CTaosInterface.consume(self._sub) + buffer = [[] for i in range(len(fields))] + while True: + block, num_of_fields = CTaosInterface.fetchBlock(result, fields) + if num_of_fields == 0: break + for i in range(len(fields)): + buffer[i].extend(block[i]) + + self.fields = fields + return list(map(tuple, zip(*buffer))) + + + def close(self, keepProgress = True): + """Close the Subscription. + """ + if self._sub is None: + return False + + CTaosInterface.unsubscribe(self._sub, keepProgress) + return True + + +if __name__ == '__main__': + from .connection import TDengineConnection + conn = TDengineConnection(host="127.0.0.1", user="root", password="taosdata", database="test") + + # Generate a cursor object to run SQL commands + sub = conn.subscribe(True, "test", "select * from meters;", 1000) + + for i in range(0,10): + data = sub.consume() + for d in data: + print(d) + + sub.close() + conn.close() \ No newline at end of file diff --git a/src/connector/python/linux/python3/taos/cinterface.py b/src/connector/python/linux/python3/taos/cinterface.py index 259c8bbd060b44f7c1b60b5c015519ed862c8ec2..7fcedc9fe9400cc8db007897906d4568c2eb234f 100644 --- a/src/connector/python/linux/python3/taos/cinterface.py +++ b/src/connector/python/linux/python3/taos/cinterface.py @@ -144,6 +144,8 @@ class CTaosInterface(object): libtaos.taos_use_result.restype = ctypes.c_void_p libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) libtaos.taos_errstr.restype = ctypes.c_char_p + libtaos.taos_subscribe.restype = ctypes.c_void_p + libtaos.taos_consume.restype = ctypes.c_void_p def __init__(self, config=None): ''' @@ -252,6 +254,41 @@ class CTaosInterface(object): """ return CTaosInterface.libtaos.taos_affected_rows(connection) + @staticmethod + def subscribe(connection, restart, topic, sql, interval): + """Create a subscription + @restart boolean, + @sql string, sql statement for data query, must be a 'select' statement. + @topic string, name of this subscription + """ + return ctypes.c_void_p(CTaosInterface.libtaos.taos_subscribe( + connection, + 1 if restart else 0, + ctypes.c_char_p(topic.encode('utf-8')), + ctypes.c_char_p(sql.encode('utf-8')), + None, + None, + interval)) + + @staticmethod + def consume(sub): + """Consume data of a subscription + """ + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_consume(sub)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.libtaos.taos_num_fields(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + return result, fields + + @staticmethod + def unsubscribe(sub, keepProgress): + """Cancel a subscription + """ + CTaosInterface.libtaos.taos_unsubscribe(sub, 1 if keepProgress else 0) + @staticmethod def useResult(connection): '''Use result after calling self.query diff --git a/src/connector/python/linux/python3/taos/connection.py b/src/connector/python/linux/python3/taos/connection.py index ba24209552600d6ee75258f929eeff829dd7b675..04fbbdec04144624a0b1f4ba25083a91ade21cce 100644 --- a/src/connector/python/linux/python3/taos/connection.py +++ b/src/connector/python/linux/python3/taos/connection.py @@ -1,5 +1,5 @@ -# from .cursor import TDengineCursor from .cursor import TDengineCursor +from .subscription import TDengineSubscription from .cinterface import CTaosInterface class TDengineConnection(object): @@ -50,6 +50,14 @@ class TDengineConnection(object): """ return CTaosInterface.close(self._conn) + def subscribe(self, restart, topic, sql, interval): + """Create a subscription. + """ + if self._conn is None: + return None + sub = CTaosInterface.subscribe(self._conn, restart, topic, sql, interval) + return TDengineSubscription(sub) + def cursor(self): """Return a new Cursor object using the connection. """ diff --git a/src/connector/python/linux/python3/taos/subscription.py b/src/connector/python/linux/python3/taos/subscription.py new file mode 100644 index 0000000000000000000000000000000000000000..d3cf10d5ada578687689b94454378dd543368e3e --- /dev/null +++ b/src/connector/python/linux/python3/taos/subscription.py @@ -0,0 +1,52 @@ +from .cinterface import CTaosInterface +from .error import * + +class TDengineSubscription(object): + """TDengine subscription object + """ + def __init__(self, sub): + self._sub = sub + + + def consume(self): + """Consume rows of a subscription + """ + if self._sub is None: + raise OperationalError("Invalid use of consume") + + result, fields = CTaosInterface.consume(self._sub) + buffer = [[] for i in range(len(fields))] + while True: + block, num_of_fields = CTaosInterface.fetchBlock(result, fields) + if num_of_fields == 0: break + for i in range(len(fields)): + buffer[i].extend(block[i]) + + self.fields = fields + return list(map(tuple, zip(*buffer))) + + + def close(self, keepProgress = True): + """Close the Subscription. + """ + if self._sub is None: + return False + + CTaosInterface.unsubscribe(self._sub, keepProgress) + return True + + +if __name__ == '__main__': + from .connection import TDengineConnection + conn = TDengineConnection(host="127.0.0.1", user="root", password="taosdata", database="test") + + # Generate a cursor object to run SQL commands + sub = conn.subscribe(True, "test", "select * from meters;", 1000) + + for i in range(0,10): + data = sub.consume() + for d in data: + print(d) + + sub.close() + conn.close() \ No newline at end of file diff --git a/src/connector/python/windows/python2/taos/cinterface.py b/src/connector/python/windows/python2/taos/cinterface.py index 8e3b7019290153a8bce475a5f2db43bc8ab04399..f8cdfcc51ea1ea9ae5789c47f2b9e54879a53934 100644 --- a/src/connector/python/windows/python2/taos/cinterface.py +++ b/src/connector/python/windows/python2/taos/cinterface.py @@ -13,14 +13,14 @@ def _convert_microsecond_to_datetime(micro): def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False): """Function to convert C bool row to python row """ - _timstamp_converter = _convert_millisecond_to_datetime + _timestamp_converter = _convert_millisecond_to_datetime if micro: - _timstamp_converter = _convert_microsecond_to_datetime + _timestamp_converter = _convert_microsecond_to_datetime if num_of_rows > 0: - return list(map(_timstamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1])) + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1])) else: - return list(map(_timstamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)])) + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)])) def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False): """Function to convert C bool row to python row @@ -144,6 +144,8 @@ class CTaosInterface(object): libtaos.taos_use_result.restype = ctypes.c_void_p libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) libtaos.taos_errstr.restype = ctypes.c_char_p + libtaos.taos_subscribe.restype = ctypes.c_void_p + libtaos.taos_consume.restype = ctypes.c_void_p def __init__(self, config=None): ''' @@ -252,6 +254,41 @@ class CTaosInterface(object): """ return CTaosInterface.libtaos.taos_affected_rows(connection) + @staticmethod + def subscribe(connection, restart, topic, sql, interval): + """Create a subscription + @restart boolean, + @sql string, sql statement for data query, must be a 'select' statement. + @topic string, name of this subscription + """ + return ctypes.c_void_p(CTaosInterface.libtaos.taos_subscribe( + connection, + 1 if restart else 0, + ctypes.c_char_p(topic.encode('utf-8')), + ctypes.c_char_p(sql.encode('utf-8')), + None, + None, + interval)) + + @staticmethod + def consume(sub): + """Consume data of a subscription + """ + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_consume(sub)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.libtaos.taos_num_fields(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + return result, fields + + @staticmethod + def unsubscribe(sub, keepProgress): + """Cancel a subscription + """ + CTaosInterface.libtaos.taos_unsubscribe(sub, 1 if keepProgress else 0) + @staticmethod def useResult(connection): '''Use result after calling self.query @@ -275,8 +312,8 @@ class CTaosInterface(object): if num_of_rows == 0: return None, 0 - blocks = [None] * len(fields) isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) + blocks = [None] * len(fields) for i in range(len(fields)): data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] @@ -351,4 +388,20 @@ class CTaosInterface(object): def errStr(connection): """Return the error styring """ - return CTaosInterface.libtaos.taos_errstr(connection) \ No newline at end of file + return CTaosInterface.libtaos.taos_errstr(connection) + + +if __name__ == '__main__': + cinter = CTaosInterface() + conn = cinter.connect() + + print('Query return value: {}'.format(cinter.query(conn, 'show databases'))) + print('Affected rows: {}'.format(cinter.affectedRows(conn))) + + result, des = CTaosInterface.useResult(conn) + + data, num_of_rows = CTaosInterface.fetchBlock(result, des) + + print(data) + + cinter.close(conn) \ No newline at end of file diff --git a/src/connector/python/windows/python2/taos/connection.py b/src/connector/python/windows/python2/taos/connection.py index ba24209552600d6ee75258f929eeff829dd7b675..e2783975d9c0f63c82a90b41a11fbaa0a3ffb5ac 100644 --- a/src/connector/python/windows/python2/taos/connection.py +++ b/src/connector/python/windows/python2/taos/connection.py @@ -1,5 +1,5 @@ -# from .cursor import TDengineCursor from .cursor import TDengineCursor +from .subscription import TDengineSubscription from .cinterface import CTaosInterface class TDengineConnection(object): @@ -15,7 +15,8 @@ class TDengineConnection(object): self._config = None self._chandle = None - self.config(**kwargs) + if len(kwargs) > 0: + self.config(**kwargs) def config(self, **kwargs): # host @@ -50,6 +51,14 @@ class TDengineConnection(object): """ return CTaosInterface.close(self._conn) + def subscribe(self, restart, topic, sql, interval): + """Create a subscription. + """ + if self._conn is None: + return None + sub = CTaosInterface.subscribe(self._conn, restart, topic, sql, interval) + return TDengineSubscription(sub) + def cursor(self): """Return a new Cursor object using the connection. """ diff --git a/src/connector/python/windows/python2/taos/subscription.py b/src/connector/python/windows/python2/taos/subscription.py new file mode 100644 index 0000000000000000000000000000000000000000..d3cf10d5ada578687689b94454378dd543368e3e --- /dev/null +++ b/src/connector/python/windows/python2/taos/subscription.py @@ -0,0 +1,52 @@ +from .cinterface import CTaosInterface +from .error import * + +class TDengineSubscription(object): + """TDengine subscription object + """ + def __init__(self, sub): + self._sub = sub + + + def consume(self): + """Consume rows of a subscription + """ + if self._sub is None: + raise OperationalError("Invalid use of consume") + + result, fields = CTaosInterface.consume(self._sub) + buffer = [[] for i in range(len(fields))] + while True: + block, num_of_fields = CTaosInterface.fetchBlock(result, fields) + if num_of_fields == 0: break + for i in range(len(fields)): + buffer[i].extend(block[i]) + + self.fields = fields + return list(map(tuple, zip(*buffer))) + + + def close(self, keepProgress = True): + """Close the Subscription. + """ + if self._sub is None: + return False + + CTaosInterface.unsubscribe(self._sub, keepProgress) + return True + + +if __name__ == '__main__': + from .connection import TDengineConnection + conn = TDengineConnection(host="127.0.0.1", user="root", password="taosdata", database="test") + + # Generate a cursor object to run SQL commands + sub = conn.subscribe(True, "test", "select * from meters;", 1000) + + for i in range(0,10): + data = sub.consume() + for d in data: + print(d) + + sub.close() + conn.close() \ No newline at end of file diff --git a/src/connector/python/windows/python3/taos/cinterface.py b/src/connector/python/windows/python3/taos/cinterface.py index 2cddf5fccf606be3d6dc60a538636204ee59565c..b4b44e199c37cf90c9beddb16433591bc0713b23 100644 --- a/src/connector/python/windows/python3/taos/cinterface.py +++ b/src/connector/python/windows/python3/taos/cinterface.py @@ -1,370 +1,407 @@ -import ctypes -from .constants import FieldType -from .error import * -import math -import datetime - -def _convert_millisecond_to_datetime(milli): - return datetime.datetime.fromtimestamp(milli/1000.0) - -def _convert_microsecond_to_datetime(micro): - return datetime.datetime.fromtimestamp(micro/1000000.0) - -def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C bool row to python row - """ - _timestamp_converter = _convert_millisecond_to_datetime - if micro: - _timestamp_converter = _convert_microsecond_to_datetime - - if num_of_rows > 0: - return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1])) - else: - return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)])) - -def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C bool row to python row - """ - if num_of_rows > 0: - return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)][::-1] ] - else: - return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[:abs(num_of_rows)] ] - -def _crow_tinyint_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C tinyint row to python row - """ - if num_of_rows > 0: - return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)][::-1] ] - else: - return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)] ] - -def _crow_smallint_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C smallint row to python row - """ - if num_of_rows > 0: - return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)][::-1]] - else: - return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)] ] - -def _crow_int_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C int row to python row - """ - if num_of_rows > 0: - return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)][::-1] ] - else: - return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)] ] - -def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C bigint row to python row - """ - if num_of_rows > 0: - return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1] ] - else: - return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)] ] - -def _crow_float_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C float row to python row - """ - if num_of_rows > 0: - return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)][::-1] ] - else: - return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)] ] - -def _crow_double_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C double row to python row - """ - if num_of_rows > 0: - return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)][::-1] ] - else: - return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)] ] - -def _crow_binary_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C binary row to python row - """ - if num_of_rows > 0: - return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)][::-1]] - else: - return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)]] - -def _crow_nchar_to_python(data, num_of_rows, nbytes=None, micro=False): - """Function to convert C nchar row to python row - """ - assert(nbytes is not None) - - res = [] - - for i in range(abs(num_of_rows)): - try: - if num_of_rows >= 0: - res.append( (ctypes.cast(data+nbytes*(abs(num_of_rows - i -1)), ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) - else: - res.append( (ctypes.cast(data+nbytes*i, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) - except ValueError: - res.append(None) - - return res - # if num_of_rows > 0: - # for i in range(abs(num_of_rows)): - # try: - # res.append( (ctypes.cast(data+nbytes*i, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) - # except ValueError: - # res.append(None) - # return res - # # return [ele.value for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[:abs(num_of_rows)][::-1]] - # else: - # return [ele.value for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[:abs(num_of_rows)]] - -_CONVERT_FUNC = { - FieldType.C_BOOL: _crow_bool_to_python, - FieldType.C_TINYINT : _crow_tinyint_to_python, - FieldType.C_SMALLINT : _crow_smallint_to_python, - FieldType.C_INT : _crow_int_to_python, - FieldType.C_BIGINT : _crow_bigint_to_python, - FieldType.C_FLOAT : _crow_float_to_python, - FieldType.C_DOUBLE : _crow_double_to_python, - FieldType.C_BINARY: _crow_binary_to_python, - FieldType.C_TIMESTAMP : _crow_timestamp_to_python, - FieldType.C_NCHAR : _crow_nchar_to_python -} - -# Corresponding TAOS_FIELD structure in C -class TaosField(ctypes.Structure): - _fields_ = [('name', ctypes.c_char * 64), - ('bytes', ctypes.c_short), - ('type', ctypes.c_char)] - -# C interface class -class CTaosInterface(object): - - libtaos = ctypes.windll.LoadLibrary('taos') - - libtaos.taos_fetch_fields.restype = ctypes.POINTER(TaosField) - libtaos.taos_init.restype = None - libtaos.taos_connect.restype = ctypes.c_void_p - libtaos.taos_use_result.restype = ctypes.c_void_p - libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) - libtaos.taos_errstr.restype = ctypes.c_char_p - - def __init__(self, config=None): - ''' - Function to initialize the class - @host : str, hostname to connect - @user : str, username to connect to server - @password : str, password to connect to server - @db : str, default db to use when log in - @config : str, config directory - - @rtype : None - ''' - if config is None: - self._config = ctypes.c_char_p(None) - else: - try: - self._config = ctypes.c_char_p(config.encode('utf-8')) - except AttributeError: - raise AttributeError("config is expected as a str") - - if config != None: - CTaosInterface.libtaos.taos_options(3, self._config) - - CTaosInterface.libtaos.taos_init() - - @property - def config(self): - """ Get current config - """ - return self._config - - def connect(self, host=None, user="root", password="taosdata", db=None, port=0): - ''' - Function to connect to server - - @rtype: c_void_p, TDengine handle - ''' - # host - try: - _host = ctypes.c_char_p(host.encode( - "utf-8")) if host != None else ctypes.c_char_p(None) - except AttributeError: - raise AttributeError("host is expected as a str") - - # user - try: - _user = ctypes.c_char_p(user.encode("utf-8")) - except AttributeError: - raise AttributeError("user is expected as a str") - - # password - try: - _password = ctypes.c_char_p(password.encode("utf-8")) - except AttributeError: - raise AttributeError("password is expected as a str") - - # db - try: - _db = ctypes.c_char_p( - db.encode("utf-8")) if db != None else ctypes.c_char_p(None) - except AttributeError: - raise AttributeError("db is expected as a str") - - # port - try: - _port = ctypes.c_int(port) - except TypeError: - raise TypeError("port is expected as an int") - - connection = ctypes.c_void_p(CTaosInterface.libtaos.taos_connect( - _host, _user, _password, _db, _port)) - - if connection.value == None: - print('connect to TDengine failed') - # sys.exit(1) - else: - print('connect to TDengine success') - - return connection - - @staticmethod - def close(connection): - '''Close the TDengine handle - ''' - CTaosInterface.libtaos.taos_close(connection) - print('connection is closed') - - @staticmethod - def query(connection, sql): - '''Run SQL - - @sql: str, sql string to run - - @rtype: 0 on success and -1 on failure - ''' - try: - return CTaosInterface.libtaos.taos_query(connection, ctypes.c_char_p(sql.encode('utf-8'))) - except AttributeError: - raise AttributeError("sql is expected as a string") - # finally: - # CTaosInterface.libtaos.close(connection) - - @staticmethod - def affectedRows(connection): - """The affected rows after runing query - """ - return CTaosInterface.libtaos.taos_affected_rows(connection) - - @staticmethod - def useResult(connection): - '''Use result after calling self.query - ''' - result = ctypes.c_void_p(CTaosInterface.libtaos.taos_use_result(connection)) - fields = [] - pfields = CTaosInterface.fetchFields(result) - for i in range(CTaosInterface.fieldsCount(connection)): - fields.append({'name': pfields[i].name.decode('utf-8'), - 'bytes': pfields[i].bytes, - 'type': ord(pfields[i].type)}) - - return result, fields - - @staticmethod - def fetchBlock(result, fields): - pblock = ctypes.c_void_p(0) - num_of_rows = CTaosInterface.libtaos.taos_fetch_block( - result, ctypes.byref(pblock)) - - if num_of_rows == 0: - return None, 0 - - isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) - blocks = [None] * len(fields) - for i in range(len(fields)): - data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] - - if fields[i]['type'] not in _CONVERT_FUNC: - raise DatabaseError("Invalid data type returned from database") - - blocks[i] = _CONVERT_FUNC[fields[i]['type']](data, num_of_rows, fields[i]['bytes'], isMicro) - - return blocks, abs(num_of_rows) - - @staticmethod - def freeResult(result): - CTaosInterface.libtaos.taos_free_result(result) - result.value = None - - @staticmethod - def fieldsCount(connection): - return CTaosInterface.libtaos.taos_field_count(connection) - - @staticmethod - def fetchFields(result): - return CTaosInterface.libtaos.taos_fetch_fields(result) - - # @staticmethod - # def fetchRow(result, fields): - # l = [] - # row = CTaosInterface.libtaos.taos_fetch_row(result) - # if not row: - # return None - - # for i in range(len(fields)): - # l.append(CTaosInterface.getDataValue( - # row[i], fields[i]['type'], fields[i]['bytes'])) - - # return tuple(l) - - # @staticmethod - # def getDataValue(data, dtype, byte): - # ''' - # ''' - # if not data: - # return None - - # if (dtype == CTaosInterface.TSDB_DATA_TYPE_BOOL): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TINYINT): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_SMALLINT): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY): - # return (ctypes.cast(data, ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00') - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP): - # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] - # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR): - # return (ctypes.cast(data, ctypes.c_char_p).value).rstrip('\x00') - - @staticmethod - def errno(connection): - """Return the error number. - """ - return CTaosInterface.libtaos.taos_errno(connection) - - @staticmethod - def errStr(connection): - """Return the error styring - """ - return CTaosInterface.libtaos.taos_errstr(connection).decode('utf-8') - - -if __name__ == '__main__': - cinter = CTaosInterface() - conn = cinter.connect() - - print('Query return value: {}'.format(cinter.query(conn, 'show databases'))) - print('Affected rows: {}'.format(cinter.affectedRows(conn))) - - result, des = CTaosInterface.useResult(conn) - - data, num_of_rows = CTaosInterface.fetchBlock(result, des) - - print(data) - +import ctypes +from .constants import FieldType +from .error import * +import math +import datetime + +def _convert_millisecond_to_datetime(milli): + return datetime.datetime.fromtimestamp(milli/1000.0) + +def _convert_microsecond_to_datetime(micro): + return datetime.datetime.fromtimestamp(micro/1000000.0) + +def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bool row to python row + """ + _timestamp_converter = _convert_millisecond_to_datetime + if micro: + _timestamp_converter = _convert_microsecond_to_datetime + + if num_of_rows > 0: + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1])) + else: + return list(map(_timestamp_converter, ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)])) + +def _crow_bool_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bool row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)][::-1] ] + else: + return [ None if ele == FieldType.C_BOOL_NULL else bool(ele) for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[:abs(num_of_rows)] ] + +def _crow_tinyint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C tinyint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)][::-1] ] + else: + return [ None if ele == FieldType.C_TINYINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[:abs(num_of_rows)] ] + +def _crow_smallint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C smallint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)][::-1]] + else: + return [ None if ele == FieldType.C_SMALLINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[:abs(num_of_rows)] ] + +def _crow_int_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C int row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)][::-1] ] + else: + return [ None if ele == FieldType.C_INT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[:abs(num_of_rows)] ] + +def _crow_bigint_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C bigint row to python row + """ + if num_of_rows > 0: + return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)][::-1] ] + else: + return [ None if ele == FieldType.C_BIGINT_NULL else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_longlong))[:abs(num_of_rows)] ] + +def _crow_float_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C float row to python row + """ + if num_of_rows > 0: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)][::-1] ] + else: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[:abs(num_of_rows)] ] + +def _crow_double_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C double row to python row + """ + if num_of_rows > 0: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)][::-1] ] + else: + return [ None if math.isnan(ele) else ele for ele in ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[:abs(num_of_rows)] ] + +def _crow_binary_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C binary row to python row + """ + if num_of_rows > 0: + return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)][::-1]] + else: + return [ None if ele.value[0:1] == FieldType.C_BINARY_NULL else ele.value.decode('utf-8') for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_char * nbytes)))[:abs(num_of_rows)]] + +def _crow_nchar_to_python(data, num_of_rows, nbytes=None, micro=False): + """Function to convert C nchar row to python row + """ + assert(nbytes is not None) + + res = [] + + for i in range(abs(num_of_rows)): + try: + if num_of_rows >= 0: + res.append( (ctypes.cast(data+nbytes*(abs(num_of_rows - i -1)), ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) + else: + res.append( (ctypes.cast(data+nbytes*i, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) + except ValueError: + res.append(None) + + return res + # if num_of_rows > 0: + # for i in range(abs(num_of_rows)): + # try: + # res.append( (ctypes.cast(data+nbytes*i, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[0].value ) + # except ValueError: + # res.append(None) + # return res + # # return [ele.value for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[:abs(num_of_rows)][::-1]] + # else: + # return [ele.value for ele in (ctypes.cast(data, ctypes.POINTER(ctypes.c_wchar * (nbytes//4))))[:abs(num_of_rows)]] + +_CONVERT_FUNC = { + FieldType.C_BOOL: _crow_bool_to_python, + FieldType.C_TINYINT : _crow_tinyint_to_python, + FieldType.C_SMALLINT : _crow_smallint_to_python, + FieldType.C_INT : _crow_int_to_python, + FieldType.C_BIGINT : _crow_bigint_to_python, + FieldType.C_FLOAT : _crow_float_to_python, + FieldType.C_DOUBLE : _crow_double_to_python, + FieldType.C_BINARY: _crow_binary_to_python, + FieldType.C_TIMESTAMP : _crow_timestamp_to_python, + FieldType.C_NCHAR : _crow_nchar_to_python +} + +# Corresponding TAOS_FIELD structure in C +class TaosField(ctypes.Structure): + _fields_ = [('name', ctypes.c_char * 64), + ('bytes', ctypes.c_short), + ('type', ctypes.c_char)] + +# C interface class +class CTaosInterface(object): + + libtaos = ctypes.windll.LoadLibrary('taos') + + libtaos.taos_fetch_fields.restype = ctypes.POINTER(TaosField) + libtaos.taos_init.restype = None + libtaos.taos_connect.restype = ctypes.c_void_p + libtaos.taos_use_result.restype = ctypes.c_void_p + libtaos.taos_fetch_row.restype = ctypes.POINTER(ctypes.c_void_p) + libtaos.taos_errstr.restype = ctypes.c_char_p + libtaos.taos_subscribe.restype = ctypes.c_void_p + libtaos.taos_consume.restype = ctypes.c_void_p + + def __init__(self, config=None): + ''' + Function to initialize the class + @host : str, hostname to connect + @user : str, username to connect to server + @password : str, password to connect to server + @db : str, default db to use when log in + @config : str, config directory + + @rtype : None + ''' + if config is None: + self._config = ctypes.c_char_p(None) + else: + try: + self._config = ctypes.c_char_p(config.encode('utf-8')) + except AttributeError: + raise AttributeError("config is expected as a str") + + if config != None: + CTaosInterface.libtaos.taos_options(3, self._config) + + CTaosInterface.libtaos.taos_init() + + @property + def config(self): + """ Get current config + """ + return self._config + + def connect(self, host=None, user="root", password="taosdata", db=None, port=0): + ''' + Function to connect to server + + @rtype: c_void_p, TDengine handle + ''' + # host + try: + _host = ctypes.c_char_p(host.encode( + "utf-8")) if host != None else ctypes.c_char_p(None) + except AttributeError: + raise AttributeError("host is expected as a str") + + # user + try: + _user = ctypes.c_char_p(user.encode("utf-8")) + except AttributeError: + raise AttributeError("user is expected as a str") + + # password + try: + _password = ctypes.c_char_p(password.encode("utf-8")) + except AttributeError: + raise AttributeError("password is expected as a str") + + # db + try: + _db = ctypes.c_char_p( + db.encode("utf-8")) if db != None else ctypes.c_char_p(None) + except AttributeError: + raise AttributeError("db is expected as a str") + + # port + try: + _port = ctypes.c_int(port) + except TypeError: + raise TypeError("port is expected as an int") + + connection = ctypes.c_void_p(CTaosInterface.libtaos.taos_connect( + _host, _user, _password, _db, _port)) + + if connection.value == None: + print('connect to TDengine failed') + # sys.exit(1) + else: + print('connect to TDengine success') + + return connection + + @staticmethod + def close(connection): + '''Close the TDengine handle + ''' + CTaosInterface.libtaos.taos_close(connection) + print('connection is closed') + + @staticmethod + def query(connection, sql): + '''Run SQL + + @sql: str, sql string to run + + @rtype: 0 on success and -1 on failure + ''' + try: + return CTaosInterface.libtaos.taos_query(connection, ctypes.c_char_p(sql.encode('utf-8'))) + except AttributeError: + raise AttributeError("sql is expected as a string") + # finally: + # CTaosInterface.libtaos.close(connection) + + @staticmethod + def affectedRows(connection): + """The affected rows after runing query + """ + return CTaosInterface.libtaos.taos_affected_rows(connection) + + @staticmethod + def subscribe(connection, restart, topic, sql, interval): + """Create a subscription + @restart boolean, + @sql string, sql statement for data query, must be a 'select' statement. + @topic string, name of this subscription + """ + return ctypes.c_void_p(CTaosInterface.libtaos.taos_subscribe( + connection, + 1 if restart else 0, + ctypes.c_char_p(topic.encode('utf-8')), + ctypes.c_char_p(sql.encode('utf-8')), + None, + None, + interval)) + + @staticmethod + def consume(sub): + """Consume data of a subscription + """ + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_consume(sub)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.libtaos.taos_num_fields(result)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + return result, fields + + @staticmethod + def unsubscribe(sub, keepProgress): + """Cancel a subscription + """ + CTaosInterface.libtaos.taos_unsubscribe(sub, 1 if keepProgress else 0) + + @staticmethod + def useResult(connection): + '''Use result after calling self.query + ''' + result = ctypes.c_void_p(CTaosInterface.libtaos.taos_use_result(connection)) + fields = [] + pfields = CTaosInterface.fetchFields(result) + for i in range(CTaosInterface.fieldsCount(connection)): + fields.append({'name': pfields[i].name.decode('utf-8'), + 'bytes': pfields[i].bytes, + 'type': ord(pfields[i].type)}) + + return result, fields + + @staticmethod + def fetchBlock(result, fields): + pblock = ctypes.c_void_p(0) + num_of_rows = CTaosInterface.libtaos.taos_fetch_block( + result, ctypes.byref(pblock)) + + if num_of_rows == 0: + return None, 0 + + isMicro = (CTaosInterface.libtaos.taos_result_precision(result) == FieldType.C_TIMESTAMP_MICRO) + blocks = [None] * len(fields) + for i in range(len(fields)): + data = ctypes.cast(pblock, ctypes.POINTER(ctypes.c_void_p))[i] + + if fields[i]['type'] not in _CONVERT_FUNC: + raise DatabaseError("Invalid data type returned from database") + + blocks[i] = _CONVERT_FUNC[fields[i]['type']](data, num_of_rows, fields[i]['bytes'], isMicro) + + return blocks, abs(num_of_rows) + + @staticmethod + def freeResult(result): + CTaosInterface.libtaos.taos_free_result(result) + result.value = None + + @staticmethod + def fieldsCount(connection): + return CTaosInterface.libtaos.taos_field_count(connection) + + @staticmethod + def fetchFields(result): + return CTaosInterface.libtaos.taos_fetch_fields(result) + + # @staticmethod + # def fetchRow(result, fields): + # l = [] + # row = CTaosInterface.libtaos.taos_fetch_row(result) + # if not row: + # return None + + # for i in range(len(fields)): + # l.append(CTaosInterface.getDataValue( + # row[i], fields[i]['type'], fields[i]['bytes'])) + + # return tuple(l) + + # @staticmethod + # def getDataValue(data, dtype, byte): + # ''' + # ''' + # if not data: + # return None + + # if (dtype == CTaosInterface.TSDB_DATA_TYPE_BOOL): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_bool))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TINYINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_byte))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_SMALLINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_short))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_INT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_int))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BIGINT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_FLOAT): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_float))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_DOUBLE): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_double))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_BINARY): + # return (ctypes.cast(data, ctypes.POINTER(ctypes.c_char))[0:byte]).rstrip('\x00') + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_TIMESTAMP): + # return ctypes.cast(data, ctypes.POINTER(ctypes.c_long))[0] + # elif (dtype == CTaosInterface.TSDB_DATA_TYPE_NCHAR): + # return (ctypes.cast(data, ctypes.c_char_p).value).rstrip('\x00') + + @staticmethod + def errno(connection): + """Return the error number. + """ + return CTaosInterface.libtaos.taos_errno(connection) + + @staticmethod + def errStr(connection): + """Return the error styring + """ + return CTaosInterface.libtaos.taos_errstr(connection).decode('utf-8') + + +if __name__ == '__main__': + cinter = CTaosInterface() + conn = cinter.connect() + + print('Query return value: {}'.format(cinter.query(conn, 'show databases'))) + print('Affected rows: {}'.format(cinter.affectedRows(conn))) + + result, des = CTaosInterface.useResult(conn) + + data, num_of_rows = CTaosInterface.fetchBlock(result, des) + + print(data) + cinter.close(conn) \ No newline at end of file diff --git a/src/connector/python/windows/python3/taos/connection.py b/src/connector/python/windows/python3/taos/connection.py index a88e25a6db6f187f2a257303189c66851bb260f6..e2783975d9c0f63c82a90b41a11fbaa0a3ffb5ac 100644 --- a/src/connector/python/windows/python3/taos/connection.py +++ b/src/connector/python/windows/python3/taos/connection.py @@ -1,81 +1,89 @@ -# from .cursor import TDengineCursor -from .cursor import TDengineCursor -from .cinterface import CTaosInterface - -class TDengineConnection(object): - """ TDengine connection object - """ - def __init__(self, *args, **kwargs): - self._conn = None - self._host = None - self._user = "root" - self._password = "taosdata" - self._database = None - self._port = 0 - self._config = None - self._chandle = None - - if len(kwargs) > 0: - self.config(**kwargs) - - def config(self, **kwargs): - # host - if 'host' in kwargs: - self._host = kwargs['host'] - - # user - if 'user' in kwargs: - self._user = kwargs['user'] - - # password - if 'password' in kwargs: - self._password = kwargs['password'] - - # database - if 'database' in kwargs: - self._database = kwargs['database'] - - # port - if 'port' in kwargs: - self._port = kwargs['port'] - - # config - if 'config' in kwargs: - self._config = kwargs['config'] - - self._chandle = CTaosInterface(self._config) - self._conn = self._chandle.connect(self._host, self._user, self._password, self._database, self._port) - - def close(self): - """Close current connection. - """ - return CTaosInterface.close(self._conn) - - def cursor(self): - """Return a new Cursor object using the connection. - """ - return TDengineCursor(self) - - def commit(self): - """Commit any pending transaction to the database. - - Since TDengine do not support transactions, the implement is void functionality. - """ - pass - - def rollback(self): - """Void functionality - """ - pass - - def clear_result_set(self): - """Clear unused result set on this connection. - """ - result = self._chandle.useResult(self._conn)[0] - if result: - self._chandle.freeResult(result) - -if __name__ == "__main__": - conn = TDengineConnection(host='192.168.1.107') - conn.close() +from .cursor import TDengineCursor +from .subscription import TDengineSubscription +from .cinterface import CTaosInterface + +class TDengineConnection(object): + """ TDengine connection object + """ + def __init__(self, *args, **kwargs): + self._conn = None + self._host = None + self._user = "root" + self._password = "taosdata" + self._database = None + self._port = 0 + self._config = None + self._chandle = None + + if len(kwargs) > 0: + self.config(**kwargs) + + def config(self, **kwargs): + # host + if 'host' in kwargs: + self._host = kwargs['host'] + + # user + if 'user' in kwargs: + self._user = kwargs['user'] + + # password + if 'password' in kwargs: + self._password = kwargs['password'] + + # database + if 'database' in kwargs: + self._database = kwargs['database'] + + # port + if 'port' in kwargs: + self._port = kwargs['port'] + + # config + if 'config' in kwargs: + self._config = kwargs['config'] + + self._chandle = CTaosInterface(self._config) + self._conn = self._chandle.connect(self._host, self._user, self._password, self._database, self._port) + + def close(self): + """Close current connection. + """ + return CTaosInterface.close(self._conn) + + def subscribe(self, restart, topic, sql, interval): + """Create a subscription. + """ + if self._conn is None: + return None + sub = CTaosInterface.subscribe(self._conn, restart, topic, sql, interval) + return TDengineSubscription(sub) + + def cursor(self): + """Return a new Cursor object using the connection. + """ + return TDengineCursor(self) + + def commit(self): + """Commit any pending transaction to the database. + + Since TDengine do not support transactions, the implement is void functionality. + """ + pass + + def rollback(self): + """Void functionality + """ + pass + + def clear_result_set(self): + """Clear unused result set on this connection. + """ + result = self._chandle.useResult(self._conn)[0] + if result: + self._chandle.freeResult(result) + +if __name__ == "__main__": + conn = TDengineConnection(host='192.168.1.107') + conn.close() print("Hello world") \ No newline at end of file diff --git a/src/connector/python/windows/python3/taos/subscription.py b/src/connector/python/windows/python3/taos/subscription.py new file mode 100644 index 0000000000000000000000000000000000000000..d3cf10d5ada578687689b94454378dd543368e3e --- /dev/null +++ b/src/connector/python/windows/python3/taos/subscription.py @@ -0,0 +1,52 @@ +from .cinterface import CTaosInterface +from .error import * + +class TDengineSubscription(object): + """TDengine subscription object + """ + def __init__(self, sub): + self._sub = sub + + + def consume(self): + """Consume rows of a subscription + """ + if self._sub is None: + raise OperationalError("Invalid use of consume") + + result, fields = CTaosInterface.consume(self._sub) + buffer = [[] for i in range(len(fields))] + while True: + block, num_of_fields = CTaosInterface.fetchBlock(result, fields) + if num_of_fields == 0: break + for i in range(len(fields)): + buffer[i].extend(block[i]) + + self.fields = fields + return list(map(tuple, zip(*buffer))) + + + def close(self, keepProgress = True): + """Close the Subscription. + """ + if self._sub is None: + return False + + CTaosInterface.unsubscribe(self._sub, keepProgress) + return True + + +if __name__ == '__main__': + from .connection import TDengineConnection + conn = TDengineConnection(host="127.0.0.1", user="root", password="taosdata", database="test") + + # Generate a cursor object to run SQL commands + sub = conn.subscribe(True, "test", "select * from meters;", 1000) + + for i in range(0,10): + data = sub.consume() + for d in data: + print(d) + + sub.close() + conn.close() \ No newline at end of file diff --git a/src/inc/taos.h b/src/inc/taos.h index c56d0e86d73c4d9b1b51e4e2e9c487081295dd3d..d9db79fbcb74e6935ab052c702b72129cd4cc132 100644 --- a/src/inc/taos.h +++ b/src/inc/taos.h @@ -116,11 +116,10 @@ DLL_EXPORT void taos_query_a(TAOS *taos, const char *sql, void (*fp)(void *param DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, void (*fp)(void *param, TAOS_RES *, int numOfRows), void *param); DLL_EXPORT void taos_fetch_row_a(TAOS_RES *res, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row), void *param); -DLL_EXPORT TAOS_SUB *taos_subscribe(const char *host, const char *user, const char *pass, const char *db, const char *table, int64_t time, int mseconds); -DLL_EXPORT TAOS_ROW taos_consume(TAOS_SUB *tsub); -DLL_EXPORT void taos_unsubscribe(TAOS_SUB *tsub); -DLL_EXPORT int taos_subfields_count(TAOS_SUB *tsub); -DLL_EXPORT TAOS_FIELD *taos_fetch_subfields(TAOS_SUB *tsub); +typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code); +DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS* taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval); +DLL_EXPORT TAOS_RES *taos_consume(TAOS_SUB *tsub); +DLL_EXPORT void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress); DLL_EXPORT TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sql, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row), int64_t stime, void *param, void (*callback)(void *)); diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h index 3b4db01755655f81482c0685e2ea694336976203..883906b617053af79dd7292246a315a5fe13f8b7 100644 --- a/src/inc/taosmsg.h +++ b/src/inc/taosmsg.h @@ -490,6 +490,7 @@ typedef struct SColumnInfo { typedef struct SMeterSidExtInfo { int32_t sid; int64_t uid; + TSKEY key; // key for subscription char tags[]; } SMeterSidExtInfo; diff --git a/src/system/detail/src/mgmtMeter.c b/src/system/detail/src/mgmtMeter.c index 76867a0d07f3f876afcb844d246a5a14365a25bd..a2a6ed8a7d3e506ddfcd1926683a0d823906c36a 100644 --- a/src/system/detail/src/mgmtMeter.c +++ b/src/system/detail/src/mgmtMeter.c @@ -1186,6 +1186,8 @@ int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pMetr int32_t * tagLen = calloc(1, sizeof(int32_t) * pMetricMetaMsg->numOfMeters); if (result == NULL || tagLen == NULL) { + tfree(result); + tfree(tagLen); return -1; } diff --git a/src/system/detail/src/vnodeQueryImpl.c b/src/system/detail/src/vnodeQueryImpl.c index d66bfbffea59a4bb47f2e0f815090d7382deae94..ce38ada8bfef01d63e5c0d36799544558b28a967 100644 --- a/src/system/detail/src/vnodeQueryImpl.c +++ b/src/system/detail/src/vnodeQueryImpl.c @@ -13,9 +13,9 @@ * along with this program. If not, see . */ +#include "os.h" #include "hash.h" #include "hashutil.h" -#include "os.h" #include "taosmsg.h" #include "textbuffer.h" #include "ttime.h" @@ -5072,10 +5072,11 @@ static void doSetTagValueInParam(tTagSchema *pTagSchema, int32_t tagColIdx, SMet SSchema *pCol = &pTagSchema->pSchema[tagColIdx]; tVariantDestroy(param); - tVariantCreateFromBinary(param, pStr, pCol->bytes, pCol->type); if (isNull(pStr, pCol->type)) { param->nType = TSDB_DATA_TYPE_NULL; + } else { + tVariantCreateFromBinary(param, pStr, pCol->bytes, pCol->type); } } @@ -6561,12 +6562,12 @@ int32_t getDataBlocksForMeters(SMeterQuerySupportObj *pSupporter, SQuery *pQuery int32_t ret = validateCompBlockInfoSegment(pQInfo, filePath, pMeterObj->vnode, &compInfo, pMeterDataInfo[j]->offsetInHeaderFile); if (ret != TSDB_CODE_SUCCESS) { // file corrupted - clearAllMeterDataBlockInfo(pMeterDataInfo, 0, j); + clearAllMeterDataBlockInfo(pMeterDataInfo, 0, numOfMeters); return TSDB_CODE_FILE_CORRUPTED; } if (compInfo.numOfBlocks <= 0 || compInfo.uid != pMeterDataInfo[j]->pMeterObj->uid) { - clearAllMeterDataBlockInfo(pMeterDataInfo, 0, j); + clearAllMeterDataBlockInfo(pMeterDataInfo, 0, numOfMeters); continue; } @@ -6574,10 +6575,13 @@ int32_t getDataBlocksForMeters(SMeterQuerySupportObj *pSupporter, SQuery *pQuery size_t bufferSize = size + sizeof(TSCKSUM); pMeterDataInfo[j]->numOfBlocks = compInfo.numOfBlocks; - pMeterDataInfo[j]->pBlock = calloc(1, bufferSize); - if (pMeterDataInfo[j]->pBlock == NULL) { - clearAllMeterDataBlockInfo(pMeterDataInfo, 0, j); + char* p = realloc(pMeterDataInfo[j]->pBlock, bufferSize); + if (p == NULL) { + clearAllMeterDataBlockInfo(pMeterDataInfo, 0, numOfMeters); return TSDB_CODE_SERV_OUT_OF_MEMORY; + } else { + memset(p, 0, bufferSize); + pMeterDataInfo[j]->pBlock = (SCompBlock*) p; } read(pVnodeFileInfo->headerFd, pMeterDataInfo[j]->pBlock, bufferSize); @@ -6589,7 +6593,7 @@ int32_t getDataBlocksForMeters(SMeterQuerySupportObj *pSupporter, SQuery *pQuery ret = validateCompBlockSegment(pQInfo, filePath, &compInfo, (char *)pMeterDataInfo[j]->pBlock, pMeterObj->vnode, checksum); if (ret != TSDB_CODE_SUCCESS) { - clearAllMeterDataBlockInfo(pMeterDataInfo, 0, j); + clearAllMeterDataBlockInfo(pMeterDataInfo, 0, numOfMeters); return TSDB_CODE_FILE_CORRUPTED; } @@ -6613,8 +6617,8 @@ int32_t getDataBlocksForMeters(SMeterQuerySupportObj *pSupporter, SQuery *pQuery } if (!setValidDataBlocks(pMeterDataInfo[j], end)) { - clearAllMeterDataBlockInfo(pMeterDataInfo, 0, j); - + clearAllMeterDataBlockInfo(pMeterDataInfo, 0, numOfMeters); + pQInfo->killed = 1; // set query kill, abort current query since no memory available return TSDB_CODE_SERV_OUT_OF_MEMORY; } diff --git a/src/system/detail/src/vnodeQueryProcess.c b/src/system/detail/src/vnodeQueryProcess.c index ebfc246f5413f82450faee946a85a57e947e1bc6..6fe8b2fa775b57799b150f57ba3f0eb71e7ab3f6 100644 --- a/src/system/detail/src/vnodeQueryProcess.c +++ b/src/system/detail/src/vnodeQueryProcess.c @@ -96,7 +96,7 @@ static void enableExecutionForNextTable(SQueryRuntimeEnv *pRuntimeEnv) { } } -static SMeterDataInfo *queryOnMultiDataCache(SQInfo *pQInfo, SMeterDataInfo *pMeterInfo) { +static void queryOnMultiDataCache(SQInfo *pQInfo, SMeterDataInfo *pMeterInfo) { SQuery * pQuery = &pQInfo->query; SMeterQuerySupportObj *pSupporter = pQInfo->pMeterQuerySupporter; SQueryRuntimeEnv * pRuntimeEnv = &pQInfo->pMeterQuerySupporter->runtimeEnv; @@ -118,7 +118,7 @@ static SMeterDataInfo *queryOnMultiDataCache(SQInfo *pQInfo, SMeterDataInfo *pMe int32_t end = pSupporter->pSidSet->starterPos[groupIdx + 1] - 1; if (isQueryKilled(pQuery)) { - return pMeterInfo; + return; } for (int32_t k = start; k <= end; ++k) { @@ -171,7 +171,7 @@ static SMeterDataInfo *queryOnMultiDataCache(SQInfo *pQInfo, SMeterDataInfo *pMe int32_t ret = setIntervalQueryExecutionContext(pSupporter, k, pMeterQueryInfo); if (ret != TSDB_CODE_SUCCESS) { pQInfo->killed = 1; - return NULL; + return; } } @@ -262,11 +262,9 @@ static SMeterDataInfo *queryOnMultiDataCache(SQInfo *pQInfo, SMeterDataInfo *pMe dTrace("QInfo:%p complete check %d cache blocks, elapsed time:%.3fms", pQInfo, totalBlocks, time / 1000.0); setQueryStatus(pQuery, QUERY_NOT_COMPLETED); - - return pMeterInfo; } -static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMeterDataInfo) { +static void queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMeterDataInfo) { SQuery * pQuery = &pQInfo->query; SMeterQuerySupportObj *pSupporter = pQInfo->pMeterQuerySupporter; SQueryRuntimeEnv * pRuntimeEnv = &pSupporter->runtimeEnv; @@ -320,7 +318,7 @@ static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMe pQInfo->code = -ret; pQInfo->killed = 1; - return NULL; + return; } dTrace("QInfo:%p file:%s, %d meters qualified", pQInfo, pVnodeFileInfo->dataFilePath, numOfQualifiedMeters); @@ -342,7 +340,7 @@ static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMe pQInfo->code = -ret; pQInfo->killed = 1; - return NULL; + return; } dTrace("QInfo:%p file:%s, %d meters contains %d blocks to be checked", pQInfo, pVnodeFileInfo->dataFilePath, @@ -362,7 +360,7 @@ static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMe pQInfo->code = -ret; pQInfo->killed = 1; - return NULL; + return; } dTrace("QInfo:%p start to load %d blocks and check", pQInfo, numOfBlocks); @@ -419,7 +417,7 @@ static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMe if (ret != TSDB_CODE_SUCCESS) { tfree(pReqMeterDataInfo); // error code has been set pQInfo->killed = 1; - return NULL; + return; } } @@ -468,8 +466,6 @@ static SMeterDataInfo *queryOnMultiDataFiles(SQInfo *pQInfo, SMeterDataInfo *pMe setQueryStatus(pQuery, QUERY_NOT_COMPLETED); freeMeterBlockInfoEx(pDataBlockInfoEx, nAllocBlocksInfoSize); - - return pMeterDataInfo; } static bool multimeterMultioutputHelper(SQInfo *pQInfo, bool *dataInDisk, bool *dataInCache, int32_t index, @@ -594,9 +590,9 @@ static void vnodeSTableSeqProcessor(SQInfo *pQInfo) { int32_t vid = getMeterObj(pSupporter->pMetersHashTable, pMeterSidExtInfo[0]->sid)->vnode; - resetCtxOutputBuf(pRuntimeEnv); - if (isPointInterpoQuery(pQuery)) { + resetCtxOutputBuf(pRuntimeEnv); + assert(pQuery->limit.offset == 0 && pQuery->limit.limit != 0); while (pSupporter->subgroupIdx < pSids->numOfSubSet) { @@ -686,6 +682,7 @@ static void vnodeSTableSeqProcessor(SQInfo *pQInfo) { return; } + resetCtxOutputBuf(pRuntimeEnv); resetSlidingWindowInfo(&pRuntimeEnv->swindowResInfo, pQuery->numOfOutputCols); while (pSupporter->meterIdx < pSupporter->numOfMeters) { @@ -696,6 +693,12 @@ static void vnodeSTableSeqProcessor(SQInfo *pQInfo) { return; } + + TSKEY skey = pQInfo->pMeterQuerySupporter->pMeterSidExtInfo[k]->key; + if (skey > 0) { + pQuery->skey = skey; + } + bool dataInDisk = true; bool dataInCache = true; if (!multimeterMultioutputHelper(pQInfo, &dataInDisk, &dataInCache, k, 0)) { @@ -759,6 +762,8 @@ static void vnodeSTableSeqProcessor(SQInfo *pQInfo) { pQuery->ekey = pSupporter->rawEKey; pSupporter->meterIdx++; + pQInfo->pMeterQuerySupporter->pMeterSidExtInfo[k]->key = pQuery->lastKey; + // if the buffer is full or group by each table, we need to jump out of the loop if (Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL) || isGroupbyEachTable(pQuery->pGroupbyExpr, pSupporter->pSidSet)) { @@ -773,6 +778,7 @@ static void vnodeSTableSeqProcessor(SQInfo *pQInfo) { assert(!Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)); continue; } else { + pQInfo->pMeterQuerySupporter->pMeterSidExtInfo[k]->key = pQuery->lastKey; // buffer is full, wait for the next round to retrieve data from current meter assert(Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)); break; @@ -833,19 +839,19 @@ static void doOrderedScan(SQInfo *pQInfo) { SQuery * pQuery = &pQInfo->query; if (QUERY_IS_ASC_QUERY(pQuery)) { - pSupporter->pMeterDataInfo = queryOnMultiDataFiles(pQInfo, pSupporter->pMeterDataInfo); + queryOnMultiDataFiles(pQInfo, pSupporter->pMeterDataInfo); if (pQInfo->code != TSDB_CODE_SUCCESS) { return; } - pSupporter->pMeterDataInfo = queryOnMultiDataCache(pQInfo, pSupporter->pMeterDataInfo); + queryOnMultiDataCache(pQInfo, pSupporter->pMeterDataInfo); } else { - pSupporter->pMeterDataInfo = queryOnMultiDataCache(pQInfo, pSupporter->pMeterDataInfo); + queryOnMultiDataCache(pQInfo, pSupporter->pMeterDataInfo); if (pQInfo->code != TSDB_CODE_SUCCESS) { return; } - pSupporter->pMeterDataInfo = queryOnMultiDataFiles(pQInfo, pSupporter->pMeterDataInfo); + queryOnMultiDataFiles(pQInfo, pSupporter->pMeterDataInfo); } } diff --git a/src/system/detail/src/vnodeRead.c b/src/system/detail/src/vnodeRead.c index c7658ca9ac35db64c5ebc2f23055d05821847a1c..71dd088ae97b8110c85cc3733c6500ddc52e06f4 100644 --- a/src/system/detail/src/vnodeRead.c +++ b/src/system/detail/src/vnodeRead.c @@ -619,7 +619,13 @@ void *vnodeQueryOnSingleTable(SMeterObj **pMetersObj, SSqlGroupbyExpr *pGroupbyE pQuery = &(pQInfo->query); dTrace("qmsg:%p create QInfo:%p, QInfo created", pQueryMsg, pQInfo); - pQuery->skey = pQueryMsg->skey; + SMeterSidExtInfo** pSids = (SMeterSidExtInfo**)pQueryMsg->pSidExtInfo; + if (pSids != NULL && pSids[0]->key > 0) { + pQuery->skey = pSids[0]->key; + } else { + pQuery->skey = pQueryMsg->skey; + } + pQuery->ekey = pQueryMsg->ekey; pQuery->lastKey = pQuery->skey; @@ -742,7 +748,6 @@ void *vnodeQueryOnMultiMeters(SMeterObj **pMetersObj, SSqlGroupbyExpr *pGroupbyE POINTER_BYTES); } - pSupporter->pMeterSidExtInfo = (SMeterSidExtInfo **)pQueryMsg->pSidExtInfo; int32_t sidElemLen = pQueryMsg->tagLength + sizeof(SMeterSidExtInfo); int32_t size = POINTER_BYTES * pQueryMsg->numOfSids + sidElemLen * pQueryMsg->numOfSids; @@ -756,12 +761,16 @@ void *vnodeQueryOnMultiMeters(SMeterObj **pMetersObj, SSqlGroupbyExpr *pGroupbyE char *px = ((char *)pSupporter->pMeterSidExtInfo) + POINTER_BYTES * pQueryMsg->numOfSids; for (int32_t i = 0; i < pQueryMsg->numOfSids; ++i) { - pSupporter->pMeterSidExtInfo[i] = (SMeterSidExtInfo *)px; - pSupporter->pMeterSidExtInfo[i]->sid = ((SMeterSidExtInfo **)pQueryMsg->pSidExtInfo)[i]->sid; + SMeterSidExtInfo* pSrc = ((SMeterSidExtInfo **)pQueryMsg->pSidExtInfo)[i]; + SMeterSidExtInfo* pDst = (SMeterSidExtInfo *)px; + + pSupporter->pMeterSidExtInfo[i] = pDst; + pDst->sid = pSrc->sid; + pDst->uid = pSrc->uid; + pDst->key = pSrc->key; if (pQueryMsg->tagLength > 0) { - memcpy(pSupporter->pMeterSidExtInfo[i]->tags, ((SMeterSidExtInfo **)pQueryMsg->pSidExtInfo)[i]->tags, - pQueryMsg->tagLength); + memcpy(pDst->tags, pSrc->tags, pQueryMsg->tagLength); } px += sidElemLen; } @@ -1091,11 +1100,13 @@ int32_t vnodeConvertQueryMeterMsg(SQueryMeterMsg *pQueryMsg) { pSids[0] = (SMeterSidExtInfo *)pMsg; pSids[0]->sid = htonl(pSids[0]->sid); pSids[0]->uid = htobe64(pSids[0]->uid); + pSids[0]->key = htobe64(pSids[0]->key); for (int32_t j = 1; j < pQueryMsg->numOfSids; ++j) { pSids[j] = (SMeterSidExtInfo *)((char *)pSids[j - 1] + sizeof(SMeterSidExtInfo) + pQueryMsg->tagLength); pSids[j]->sid = htonl(pSids[j]->sid); pSids[j]->uid = htobe64(pSids[j]->uid); + pSids[j]->key = htobe64(pSids[j]->key); } pMsg = (char *)pSids[pQueryMsg->numOfSids - 1]; diff --git a/src/system/detail/src/vnodeShell.c b/src/system/detail/src/vnodeShell.c index d4b9ceb77df589e8f3007f2da39a5c65d6988e6d..69d502c61828154b6ea985399e41e5039a6f9805 100644 --- a/src/system/detail/src/vnodeShell.c +++ b/src/system/detail/src/vnodeShell.c @@ -417,6 +417,7 @@ void vnodeExecuteRetrieveReq(SSchedMsg *pSched) { int code = 0; pRetrieve = (SRetrieveMeterMsg *)pMsg; + SQInfo* pQInfo = (SQInfo*)pRetrieve->qhandle; pRetrieve->free = htons(pRetrieve->free); if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE) { @@ -443,7 +444,15 @@ void vnodeExecuteRetrieveReq(SSchedMsg *pSched) { size = vnodeGetResultSize((void *)(pRetrieve->qhandle), &numOfRows); } - pStart = taosBuildRspMsgWithSize(pObj->thandle, TSDB_MSG_TYPE_RETRIEVE_RSP, size + 100); + // buffer size for progress information, including meter count, + // and for each meter, including 'uid' and 'TSKEY'. + int progressSize = 0; + if (pQInfo->pMeterQuerySupporter != NULL) + progressSize = pQInfo->pMeterQuerySupporter->numOfMeters * (sizeof(int64_t) + sizeof(TSKEY)) + sizeof(int32_t); + else if (pQInfo->pObj != NULL) + progressSize = sizeof(int64_t) + sizeof(TSKEY) + sizeof(int32_t); + + pStart = taosBuildRspMsgWithSize(pObj->thandle, TSDB_MSG_TYPE_RETRIEVE_RSP, progressSize + size + 100); if (pStart == NULL) { taosSendSimpleRsp(pObj->thandle, TSDB_MSG_TYPE_RETRIEVE_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY); goto _exit; @@ -473,6 +482,31 @@ void vnodeExecuteRetrieveReq(SSchedMsg *pSched) { } pMsg += size; + + // write the progress information of each meter to response + // this is required by subscriptions + if (pQInfo->pMeterQuerySupporter != NULL && pQInfo->pMeterQuerySupporter->pMeterSidExtInfo != NULL) { + *((int32_t*)pMsg) = htonl(pQInfo->pMeterQuerySupporter->numOfMeters); + pMsg += sizeof(int32_t); + for (int32_t i = 0; i < pQInfo->pMeterQuerySupporter->numOfMeters; i++) { + *((int64_t*)pMsg) = htobe64(pQInfo->pMeterQuerySupporter->pMeterSidExtInfo[i]->uid); + pMsg += sizeof(int64_t); + *((TSKEY*)pMsg) = htobe64(pQInfo->pMeterQuerySupporter->pMeterSidExtInfo[i]->key); + pMsg += sizeof(TSKEY); + } + } else if (pQInfo->pObj != NULL) { + *((int32_t*)pMsg) = htonl(1); + pMsg += sizeof(int32_t); + *((int64_t*)pMsg) = htobe64(pQInfo->pObj->uid); + pMsg += sizeof(int64_t); + if (pQInfo->pointsRead > 0) { + *((TSKEY*)pMsg) = htobe64(pQInfo->query.lastKey + 1); + } else { + *((TSKEY*)pMsg) = htobe64(pQInfo->query.lastKey); + } + pMsg += sizeof(TSKEY); + } + msgLen = pMsg - pStart; assert(code != TSDB_CODE_ACTION_IN_PROGRESS); diff --git a/src/util/src/version.c b/src/util/src/version.c index 86cc279f4ce69b8fd5b80cc34bd4b63681a46e24..b8b0d611decca8eb7ca3900e15ad8c20ee6ce0d4 100644 --- a/src/util/src/version.c +++ b/src/util/src/version.c @@ -1,4 +1,4 @@ -char version[64] = "1.6.5.1"; +char version[64] = "1.6.5.3"; char compatible_version[64] = "1.6.1.0"; char gitinfo[128] = "700305490a82228ec1b0244afb838bdbb9de9793"; char gitinfoOfInternal[128] = ""; diff --git a/tests/examples/c/makefile b/tests/examples/c/makefile index ac8ff21aaffc47bbf2a36386f06996b6ad13a086..0a4b8ee9d2bd00ab3daaac0c3a93497de4fd03f8 100644 --- a/tests/examples/c/makefile +++ b/tests/examples/c/makefile @@ -3,21 +3,23 @@ ROOT=./ TARGET=exe -LFLAGS = '-Wl,-rpath,/usr/local/taos/driver' -ltaos -lpthread -lm -lrt -CFLAGS = -O3 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion -Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX -msse4.2 -Wno-unused-function -D_M_X64 -std=gnu99 +LFLAGS = '-Wl,-rpath,/usr/local/taos/driver/' -ltaos -lpthread -lm -lrt +#LFLAGS = '-Wl,-rpath,/home/zbm/project/td/debug/build/lib/' -L/home/zbm/project/td/debug/build/lib -ltaos -lpthread -lm -lrt +CFLAGS = -O3 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion -Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX -msse4.2 -Wno-unused-function -D_M_X64 \ + -I/usr/local/taos/include -std=gnu99 all: $(TARGET) exe: gcc $(CFLAGS) ./asyncdemo.c -o $(ROOT)/asyncdemo $(LFLAGS) gcc $(CFLAGS) ./demo.c -o $(ROOT)/demo $(LFLAGS) + gcc $(CFLAGS) ./prepare.c -o $(ROOT)/prepare $(LFLAGS) gcc $(CFLAGS) ./stream.c -o $(ROOT)/stream $(LFLAGS) - gcc $(CFLAGS) ./subscribe.c -o $(ROOT)/subscribe $(LFLAGS) + gcc $(CFLAGS) ./subscribe.c -o $(ROOT)subscribe $(LFLAGS) clean: - rm $(ROOT)asyncdemo - rm $(ROOT)demo - rm $(ROOT)stream - rm $(ROOT)subscribe - - \ No newline at end of file + rm $(ROOT)/asyncdemo + rm $(ROOT)/demo + rm $(ROOT)/prepare + rm $(ROOT)/stream + rm $(ROOT)/subscribe diff --git a/tests/examples/c/subscribe.c b/tests/examples/c/subscribe.c index 219fa133e02f24cbafb5d446ff3a4aacac2f9c67..0bf93f6f2ddd81e715e7d9cf0b5abfd054635060 100644 --- a/tests/examples/c/subscribe.c +++ b/tests/examples/c/subscribe.c @@ -1,18 +1,3 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * This program is free software: you can use, redistribute, and/or modify - * it under the terms of the GNU Affero General Public License, version 3 - * or later ("AGPL"), as published by the Free Software Foundation. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ - // sample code for TDengine subscribe/consume API // to compile: gcc -o subscribe subscribe.c -ltaos @@ -20,43 +5,239 @@ #include #include #include // include TDengine header file +#include + +void print_result(TAOS_RES* res, int blockFetch) { + TAOS_ROW row = NULL; + int num_fields = taos_num_fields(res); + TAOS_FIELD* fields = taos_fetch_fields(res); + int nRows = 0; + + if (blockFetch) { + nRows = taos_fetch_block(res, &row); + for (int i = 0; i < nRows; i++) { + char temp[256]; + taos_print_row(temp, row + i, fields, num_fields); + puts(temp); + } + } else { + while ((row = taos_fetch_row(res))) { + char temp[256]; + taos_print_row(temp, row, fields, num_fields); + puts(temp); + nRows++; + } + } + + printf("%d rows consumed.\n", nRows); +} -int main(int argc, char *argv[]) -{ - TAOS_SUB *tsub; + +void subscribe_callback(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code) { + print_result(res, *(int*)param); +} + + +void check_row_count(int line, TAOS_RES* res, int expected) { + int actual = 0; TAOS_ROW row; - char dbname[64], table[64]; - char temp[256]; + while ((row = taos_fetch_row(res))) { + actual++; + } + if (actual != expected) { + printf("line %d: row count mismatch, expected: %d, actual: %d\n", line, expected, actual); + } else { + printf("line %d: %d rows consumed as expected\n", line, actual); + } +} - if ( argc == 1 ) { - printf("usage: %s server-ip db-name table-name \n", argv[0]); - exit(0); - } - if ( argc >= 2 ) strcpy(dbname, argv[2]); - if ( argc >= 3 ) strcpy(table, argv[3]); +void run_test(TAOS* taos) { + taos_query(taos, "drop database if exists test;"); + + usleep(100000); + taos_query(taos, "create database test tables 5;"); + usleep(100000); + taos_query(taos, "use test;"); + usleep(100000); + taos_query(taos, "create table meters(ts timestamp, a int, b binary(20)) tags(loc binary(20), area int);"); + + taos_query(taos, "insert into t0 using meters tags('beijing', 0) values('2020-01-01 00:00:00.000', 0, 'china');"); + taos_query(taos, "insert into t0 using meters tags('beijing', 0) values('2020-01-01 00:01:00.000', 0, 'china');"); + taos_query(taos, "insert into t0 using meters tags('beijing', 0) values('2020-01-01 00:02:00.000', 0, 'china');"); + taos_query(taos, "insert into t1 using meters tags('shanghai', 0) values('2020-01-01 00:00:00.000', 0, 'china');"); + taos_query(taos, "insert into t1 using meters tags('shanghai', 0) values('2020-01-01 00:01:00.000', 0, 'china');"); + taos_query(taos, "insert into t1 using meters tags('shanghai', 0) values('2020-01-01 00:02:00.000', 0, 'china');"); + taos_query(taos, "insert into t1 using meters tags('shanghai', 0) values('2020-01-01 00:03:00.000', 0, 'china');"); + taos_query(taos, "insert into t2 using meters tags('london', 0) values('2020-01-01 00:00:00.000', 0, 'UK');"); + taos_query(taos, "insert into t2 using meters tags('london', 0) values('2020-01-01 00:01:00.000', 0, 'UK');"); + taos_query(taos, "insert into t2 using meters tags('london', 0) values('2020-01-01 00:01:01.000', 0, 'UK');"); + taos_query(taos, "insert into t2 using meters tags('london', 0) values('2020-01-01 00:01:02.000', 0, 'UK');"); + taos_query(taos, "insert into t3 using meters tags('tianjin', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t4 using meters tags('wuhan', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t5 using meters tags('jinan', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t6 using meters tags('haikou', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t7 using meters tags('nanjing', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t8 using meters tags('lanzhou', 0) values('2020-01-01 00:01:02.000', 0, 'china');"); + taos_query(taos, "insert into t9 using meters tags('tokyo', 0) values('2020-01-01 00:01:02.000', 0, 'japan');"); + + // super tables subscription + + TAOS_SUB* tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0); + TAOS_RES* res = taos_consume(tsub); + check_row_count(__LINE__, res, 18); + + res = taos_consume(tsub); + check_row_count(__LINE__, res, 0); + + taos_query(taos, "insert into t0 values('2020-01-01 00:03:00.000', 0, 'china');"); + taos_query(taos, "insert into t8 values('2020-01-01 00:01:03.000', 0, 'china');"); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 2); + + taos_query(taos, "insert into t2 values('2020-01-01 00:01:02.001', 0, 'UK');"); + taos_query(taos, "insert into t1 values('2020-01-01 00:03:00.001', 0, 'UK');"); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 2); + + taos_query(taos, "insert into t1 values('2020-01-01 00:03:00.002', 0, 'china');"); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 1); - tsub = taos_subscribe(argv[1], "root", "taosdata", dbname, table, 0, 1000); - if ( tsub == NULL ) { - printf("failed to connet to db:%s\n", dbname); + // keep progress information and restart subscription + taos_unsubscribe(tsub, 1); + taos_query(taos, "insert into t0 values('2020-01-01 00:04:00.000', 0, 'china');"); + tsub = taos_subscribe(taos, 1, "test", "select * from meters;", NULL, NULL, 0); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 24); + + // keep progress information and continue previous subscription + taos_unsubscribe(tsub, 1); + tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 0); + + // don't keep progress information and continue previous subscription + taos_unsubscribe(tsub, 0); + tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 24); + + // single meter subscription + + taos_unsubscribe(tsub, 0); + tsub = taos_subscribe(taos, 0, "test", "select * from t0;", NULL, NULL, 0); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 5); + + res = taos_consume(tsub); + check_row_count(__LINE__, res, 0); + + taos_query(taos, "insert into t0 values('2020-01-01 00:04:00.001', 0, 'china');"); + res = taos_consume(tsub); + check_row_count(__LINE__, res, 1); + + taos_unsubscribe(tsub, 0); +} + + +int main(int argc, char *argv[]) { + const char* host = "127.0.0.1"; + const char* user = "root"; + const char* passwd = "taosdata"; + const char* sql = "select * from meters;"; + const char* topic = "test-multiple"; + int async = 1, restart = 0, keep = 1, test = 0, blockFetch = 0; + + for (int i = 1; i < argc; i++) { + if (strncmp(argv[i], "-h=", 3) == 0) { + host = argv[i] + 3; + continue; + } + if (strncmp(argv[i], "-u=", 3) == 0) { + user = argv[i] + 3; + continue; + } + if (strncmp(argv[i], "-p=", 3) == 0) { + passwd = argv[i] + 3; + continue; + } + if (strcmp(argv[i], "-sync") == 0) { + async = 0; + continue; + } + if (strcmp(argv[i], "-restart") == 0) { + restart = 1; + continue; + } + if (strcmp(argv[i], "-single") == 0) { + sql = "select * from t0;"; + topic = "test-single"; + continue; + } + if (strcmp(argv[i], "-nokeep") == 0) { + keep = 0; + continue; + } + if (strncmp(argv[i], "-sql=", 5) == 0) { + sql = argv[i] + 5; + topic = "test-custom"; + continue; + } + if (strcmp(argv[i], "-test") == 0) { + test = 1; + continue; + } + if (strcmp(argv[i], "-block-fetch") == 0) { + blockFetch = 1; + continue; + } + } + + // init TAOS + taos_init(); + + TAOS* taos = taos_connect(host, user, passwd, "test", 0); + if (taos == NULL) { + printf("failed to connect to db, reason:%s\n", taos_errstr(taos)); exit(1); } - TAOS_FIELD *fields = taos_fetch_subfields(tsub); - int fcount = taos_subfields_count(tsub); + if (test) { + run_test(taos); + taos_close(taos); + exit(0); + } + + TAOS_SUB* tsub = NULL; + if (async) { + // create an asynchronized subscription, the callback function will be called every 1s + tsub = taos_subscribe(taos, restart, topic, sql, subscribe_callback, &blockFetch, 1000); + } else { + // create an synchronized subscription, need to call 'taos_consume' manually + tsub = taos_subscribe(taos, restart, topic, sql, NULL, NULL, 0); + } - printf("start to retrieve data\n"); - printf("please use other taos client, insert rows into %s.%s\n", dbname, table); - while ( 1 ) { - row = taos_consume(tsub); - if ( row == NULL ) break; + if (tsub == NULL) { + printf("failed to create subscription.\n"); + exit(0); + } - taos_print_row(temp, row, fields, fcount); - printf("%s\n", temp); + if (async) { + getchar(); + } else while(1) { + TAOS_RES* res = taos_consume(tsub); + if (res == NULL) { + printf("failed to consume data."); + break; + } else { + print_result(res, blockFetch); + getchar(); + } } - taos_unsubscribe(tsub); + taos_unsubscribe(tsub, keep); + taos_close(taos); return 0; } -