Skip to content

Commit f45ccea

Browse files
ShenaMaoShenaMao
committed
release 3.0.0.0
1 parent 59cf291 commit f45ccea

File tree

91 files changed

+89225
-394
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

91 files changed

+89225
-394
lines changed

‎core/include/ConstantImp.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2469,6 +2469,7 @@ class FastFixedLengthVector : public Vector {
24692469
int getSegmentCount(INDEX size, int segmentSizeInBit) const;
24702470
//start is inclusive, but end is exclusive
24712471
virtual bool hasNullInRange(const unsigned char* buf, INDEX start, INDEX end) const { return true;}
2472+
void* getDataArray() const override {return (void*)data_;}
24722473

24732474

24742475
protected:

‎core/include/DdbPythonUtil.h‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,14 @@ namespace dolphindb {
2525
#endif
2626
#define DLOG true?ddb::DLogger::GetMinLevel():ddb::DLogger::Info
2727

28+
#define EXPARAM_DEFAULT INT_MIN
29+
2830

2931
struct HIDEVISIBILITY Preserved {
3032
// instantiation only once for frequently use
3133
bool np_above_1_20_;
32-
bool pd_above_2_00_;
34+
bool pd_above_2_0_;
35+
bool pd_above_1_2_;
3336
bool pyarrow_import_;
3437
bool has_arrow_;
3538
// modules and methods
@@ -50,6 +53,16 @@ struct HIDEVISIBILITY Preserved {
5053
py::object pdindex_;
5154
py::object pdextensiondtype_;
5255

56+
// pandas extension dtypes (use equal)
57+
py::object pdBooleanDtype_;
58+
py::object pdFloat32Dtype_;
59+
py::object pdFloat64Dtype_;
60+
py::object pdInt8Dtype_;
61+
py::object pdInt16Dtype_;
62+
py::object pdInt32Dtype_;
63+
py::object pdInt64Dtype_;
64+
py::object pdStringDtype_;
65+
5366
// numpy dtypes (instances of dtypes, use equal)
5467
py::object nparray_;
5568
py::object npmatrix_;
@@ -197,8 +210,10 @@ class EXPORT_DECL DdbPythonUtil{
197210
}
198211
};
199212

200-
static ConstantSP toDolphinDB(py::object obj, Type typeIndicator = {DT_UNK, -1}, TableChecker checker = TableChecker());
213+
static ConstantSP toDolphinDB(py::object obj, Type typeIndicator = {DT_UNK, EXPARAM_DEFAULT}, TableChecker checker = TableChecker());
201214
static py::object toPython(ConstantSP obj, bool tableFlag=false, ToPythonOption *poption = NULL);
215+
static ConstantSP toDolphinDB_Scalar(py::object obj, Type typeIndicator = {DT_UNK, EXPARAM_DEFAULT});
216+
static ConstantSP toDolphinDB_Vector(py::object obj, Type typeIndicator = {DT_UNK, EXPARAM_DEFAULT}, CHILD_VECTOR_OPTION option = CHILD_VECTOR_OPTION::DISABLE);
202217
static py::object loadPickleFile(const std::string &filepath);
203218
static void createPyVector(const ConstantSP &obj,py::object &pyObject,bool tableFlag,ToPythonOption *poption);
204219
static Preserved *preserved_;

‎core/include/DolphinDB.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -192,14 +192,14 @@ class EXPORT_DECL Guid {
192192

193193
class HIDEVISIBILITY ProtectGil{
194194
public:
195-
ProtectGil(bool release = false, const string &name = "");
195+
ProtectGil(bool release = false, const string &name = "");
196196
void acquire();
197197
~ProtectGil();
198198
private:
199199
string name_;
200200
SmartPointer<py::gil_scoped_release> pgilRelease_;
201201
bool acquired_;
202-
PyGILState_STATE gstate_;
202+
PyGILState_STATE gstate_;
203203
};
204204

205205
struct GuidHash {

‎core/include/EventHandler.h‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
#pragma once
2+
3+
#include "SmartPointer.h"
4+
#include "SysIO.h"
5+
#include "Types.h"
6+
#include "DolphinDB.h"
7+
#include <string>
8+
#include <unordered_map>
9+
#include <utility>
10+
#include <vector>
11+
namespace dolphindb {
12+
13+
struct EventSchema{
14+
std::string eventType_;
15+
std::vector<std::string> fieldNames_;
16+
std::vector<DATA_TYPE> fieldTypes_;
17+
std::vector<DATA_FORM> fieldForms_;
18+
std::vector<int> fieldExtraParams_;
19+
};
20+
21+
struct EventSchemaEx{
22+
EventSchema schema_;
23+
int timeIndex_;
24+
std::vector<int> commonKeyIndex_;
25+
};
26+
27+
class AttributeSerializer{
28+
public:
29+
AttributeSerializer(int unitLen, DATA_FORM form): unitLen_(unitLen), form_(form) {}
30+
virtual ~AttributeSerializer() = default;
31+
virtual IO_ERR serialize(const ConstantSP& attribute, DataOutputStreamSP outStream);
32+
33+
protected:
34+
int unitLen_;
35+
DATA_FORM form_;
36+
};
37+
38+
class FastArrayAttributeSerializer : public AttributeSerializer{
39+
public:
40+
FastArrayAttributeSerializer(int unitLen) : AttributeSerializer(unitLen, DF_VECTOR) {}
41+
~FastArrayAttributeSerializer() = default;
42+
virtual IO_ERR serialize(const ConstantSP& attribute, DataOutputStreamSP outStream) override;
43+
};
44+
45+
class ScalarAttributeSerializer : public AttributeSerializer {
46+
public:
47+
ScalarAttributeSerializer(int unitLen) : AttributeSerializer(unitLen, DF_SCALAR) {buf_.resize(unitLen_);}
48+
~ScalarAttributeSerializer() = default;
49+
virtual IO_ERR serialize(const ConstantSP& attribute, DataOutputStreamSP outStream) override;
50+
51+
private:
52+
std::string buf_;
53+
};
54+
55+
class StringScalarAttributeSerializer : public AttributeSerializer {
56+
public:
57+
StringScalarAttributeSerializer(bool isBlob) : AttributeSerializer(-1, DF_SCALAR), isBlob_(isBlob) {}
58+
~StringScalarAttributeSerializer() = default;
59+
virtual IO_ERR serialize(const ConstantSP& attribute, DataOutputStreamSP outStream) override;
60+
61+
private:
62+
bool isBlob_;
63+
};
64+
65+
using AttributeSerializerSP = SmartPointer<AttributeSerializer>;
66+
using EventSchemaExSP = SmartPointer<EventSchemaEx>;
67+
68+
struct EventInfo{
69+
std::vector<AttributeSerializerSP> attributeSerializers_;
70+
EventSchemaExSP eventSchema_;
71+
};
72+
73+
class EventHandler{
74+
public:
75+
EventHandler(const std::vector<EventSchema>& eventSchemas, const std::vector<std::string>& eventTimeKeys, const std::vector<std::string>& commonKeys);
76+
bool checkOutputTable(TableSP outputTable, std::string& errMsg);
77+
bool serializeEvent(const std::string& eventType, const std::vector<ConstantSP>& attributes, std::vector<ConstantSP>& serializedEvent, std::string& errMsg);
78+
bool deserializeEvent(ConstantSP obj, std::vector<std::string>& eventTypes, std::vector<std::vector<ConstantSP>>& attributes, ErrorCodeInfo& errorInfo);
79+
private:
80+
bool checkSchema(const std::vector<EventSchema>& eventSchemas, const std::vector<std::string> &expandTimeKeys, const std::vector<std::string>& commonKeys, std::string& errMsg);
81+
ConstantSP deserializeScalar(DATA_TYPE type, int extraParam, DataInputStreamSP input, IO_ERR& ret);
82+
ConstantSP deserializeFastArray(DATA_TYPE type, int extraParam, DataInputStreamSP input, IO_ERR& ret);
83+
ConstantSP deserializeAny(DATA_TYPE type, DATA_FORM form, DataInputStreamSP input, IO_ERR& ret);
84+
private:
85+
std::unordered_map<std::string, EventInfo> eventInfos_;
86+
bool isNeedEventTime_;
87+
88+
int outputColNums_; //the number of columns of the output table
89+
int commonKeySize_;
90+
};
91+
92+
class EventSender{
93+
public:
94+
EventSender(
95+
DBConnection& conn,
96+
const std::string& tableName,
97+
const std::vector<EventSchema>& eventSchema,
98+
const std::vector<std::string>& eventTimeFields = std::vector<std::string>(),
99+
const std::vector<std::string>& commonFields = std::vector<std::string>());
100+
void sendEvent(const std::string& eventType, const std::vector<ConstantSP>& attributes);
101+
102+
private:
103+
std::string insertScript_;
104+
EventHandler eventHandler_;
105+
DBConnection& conn_;
106+
};
107+
108+
}

‎core/include/Streaming.h‎

Lines changed: 86 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
#include "Concurrent.h"
77
#include "TableImp.h"
88
#include "DolphinDB.h"
9+
#include "EventHandler.h"
910
#include "Util.h"
1011
#ifdef _MSC_VER
1112
#ifdef _USRDLL
@@ -76,6 +77,7 @@ using MessageQueueSP = SmartPointer<MessageQueue>;
7677
using MessageTableQueueSP = SmartPointer<MessageTableQueue>;
7778
using MessageHandler = std::function<void(Message)>;
7879
using MessageBatchHandler = std::function<void(vector<Message>)>;
80+
using EventMessageHandler = std::function<void(const std::string&, std::vector<ConstantSP>&)>;
7981

8082
#define DEFAULT_ACTION_NAME "cppStreamingAPI"
8183
constexpr int DEFAULT_QUEUE_CAPACITY = 65536;
@@ -102,7 +104,8 @@ typedef SmartPointer<StreamDeserializer> StreamDeserializerSP;
102104

103105
struct SubscribeInfo {
104106
SubscribeInfo()
105-
: host("INVAILD"),
107+
: ID("INVALID"),
108+
host("INVAILD"),
106109
port(-1),
107110
tableName("INVALID"),
108111
actionName("INVALID"),
@@ -116,11 +119,32 @@ struct SubscribeInfo {
116119
tqueue(nullptr),
117120
userName(""),
118121
password(""),
119-
streamDeserializer(nullptr) {}
120-
explicit SubscribeInfo(const string &host, int port, const string &tableName, const string &actionName, long long offset, bool resub,
121-
const VectorSP &filter, bool msgAsTable, bool allowExists, int batchSize,
122-
const string &userName, const string &password, const StreamDeserializerSP &blobDeserializer, const bool istqueue)
123-
: host(move(host)),
122+
streamDeserializer(nullptr),
123+
currentSiteIndex(-1),
124+
isEvent_(false),
125+
resubTimeout(100),
126+
subOnce(false),
127+
lastSiteIndex(-1) {}
128+
explicit SubscribeInfo(const string &id,
129+
const string &host,
130+
int port,
131+
const string &tableName,
132+
const string &actionName,
133+
long long offset,
134+
bool resub,
135+
const VectorSP &filter,
136+
bool msgAsTable,
137+
bool allowExists,
138+
int batchSize,
139+
const string &userName,
140+
const string &password,
141+
const StreamDeserializerSP &blobDeserializer,
142+
const bool istqueue,
143+
bool isEvent,
144+
int resubTimeout,
145+
bool subOnce)
146+
: ID(move(id)),
147+
host(move(host)),
124148
port(port),
125149
tableName(move(tableName)),
126150
actionName(move(actionName)),
@@ -136,9 +160,15 @@ struct SubscribeInfo {
136160
userName(move(userName)),
137161
password(move(password)),
138162
istqueue(istqueue),
139-
streamDeserializer(blobDeserializer){
163+
streamDeserializer(blobDeserializer),
164+
currentSiteIndex(-1),
165+
isEvent_(isEvent),
166+
resubTimeout(resubTimeout),
167+
subOnce(subOnce),
168+
lastSiteIndex(-1) {
140169
}
141170

171+
string ID;
142172
string host;
143173
int port;
144174
string tableName;
@@ -158,6 +188,12 @@ struct SubscribeInfo {
158188
SocketSP socket;
159189

160190
vector<ThreadSP> handleThread;
191+
vector<pair<string, int>> availableSites;
192+
int currentSiteIndex;
193+
bool isEvent_;
194+
int resubTimeout;
195+
bool subOnce;
196+
int lastSiteIndex;
161197
void setExitFlag() {
162198
if (istqueue) {
163199
tqueue->setExitFlag();
@@ -181,6 +217,22 @@ struct SubscribeInfo {
181217
}
182218
handleThread.clear();
183219
}
220+
221+
void updateByReconnect(int currentReconnSiteIndex, const std::string &topic) {
222+
auto thisTopicLastSuccessfulNode = this->lastSiteIndex;
223+
if (this->subOnce && thisTopicLastSuccessfulNode != currentReconnSiteIndex) {
224+
// update currentSiteIndex
225+
if (thisTopicLastSuccessfulNode < currentReconnSiteIndex) {
226+
currentReconnSiteIndex--;
227+
}
228+
// update info
229+
this->availableSites.erase(this->availableSites.begin() + thisTopicLastSuccessfulNode);
230+
this->currentSiteIndex = currentReconnSiteIndex;
231+
232+
// update lastSuccessfulNode
233+
this->lastSiteIndex = currentReconnSiteIndex;
234+
}
235+
}
184236
};
185237

186238

@@ -198,13 +250,27 @@ class EXPORT_DECL StreamingClient {
198250
int64_t offset = -1, bool resubscribe = true, const VectorSP &filter = nullptr,
199251
bool msgAsTable = false, bool allowExists = false, int batchSize = 1,
200252
string userName="", string password="",
201-
const StreamDeserializerSP &blobDeserializer = nullptr, bool istqueue = false);
253+
const StreamDeserializerSP &blobDeserializer = nullptr, bool istqueue = false,
254+
const std::vector<std::string> &backupSites = std::vector<std::string>(), bool isEvent = false,
255+
int resubTimeout = 100, bool subOnce = false);
202256
void unsubscribeInternal(string host, int port, string tableName, string actionName = DEFAULT_ACTION_NAME);
203257

204258
protected:
205259
SmartPointer<StreamingClientImpl> impl_;
206260
};
207261

262+
class EventClient : public StreamingClient{
263+
public:
264+
EventClient(const std::vector<EventSchema>& eventSchemes, const std::vector<std::string>& eventTimeKeys, const std::vector<std::string>& commonKeys);
265+
ThreadSP subscribe(const string& host, int port, const EventMessageHandler &handler, const string& tableName, const string& actionName = DEFAULT_ACTION_NAME, int64_t offset = -1,
266+
bool resub = true, const string& userName="", const string& password="");
267+
void unsubscribe(const string& host, int port, const string& tableName, const string& actionName = DEFAULT_ACTION_NAME);
268+
269+
private:
270+
EventHandler eventHandler_;
271+
};
272+
273+
208274
class EXPORT_DECL ThreadedClient : public StreamingClient {
209275
public:
210276
//listeningPort > 0 : listen mode, wait for server connection
@@ -215,13 +281,17 @@ class EXPORT_DECL ThreadedClient : public StreamingClient {
215281
string actionName = DEFAULT_ACTION_NAME, int64_t offset = -1, bool resub = true,
216282
const VectorSP &filter = nullptr, bool msgAsTable = false, bool allowExists = false,
217283
string userName="", string password="",
218-
const StreamDeserializerSP &blobDeserializer = nullptr);
284+
const StreamDeserializerSP &blobDeserializer = nullptr,
285+
const std::vector<std::string> &backupSites = std::vector<std::string>(),
286+
int resubTimeout = 100, bool subOnce = false);
219287
ThreadSP subscribe(string host, int port, const MessageBatchHandler &handler, string tableName,
220288
string actionName = DEFAULT_ACTION_NAME, int64_t offset = -1, bool resub = true,
221289
const VectorSP &filter = nullptr, bool allowExists = false, int batchSize = 1,
222290
double throttle = 1,bool msgAsTable = false,
223291
string userName = "", string password = "",
224-
const StreamDeserializerSP &blobDeserializer = nullptr);
292+
const StreamDeserializerSP &blobDeserializer = nullptr,
293+
const std::vector<std::string> &backupSites = std::vector<std::string>(),
294+
int resubTimeout = 100, bool subOnce = false);
225295
size_t getQueueDepth(const ThreadSP &thread);
226296
void unsubscribe(string host, int port, string tableName, string actionName = DEFAULT_ACTION_NAME);
227297
};
@@ -236,7 +306,9 @@ class EXPORT_DECL ThreadPooledClient : public StreamingClient {
236306
string actionName, int64_t offset = -1, bool resub = true,
237307
const VectorSP &filter = nullptr, bool msgAsTable = false, bool allowExists = false,
238308
string userName = "", string password = "",
239-
const StreamDeserializerSP &blobDeserializer = nullptr);
309+
const StreamDeserializerSP &blobDeserializer = nullptr,
310+
const std::vector<std::string> &backupSites = std::vector<std::string>(),
311+
int resubTimeout = 100, bool subOnce = false);
240312
void unsubscribe(string host, int port, string tableName, string actionName = DEFAULT_ACTION_NAME);
241313
size_t getQueueDepth(const ThreadSP &thread);
242314

@@ -254,7 +326,9 @@ class EXPORT_DECL PollingClient : public StreamingClient {
254326
int64_t offset = -1, bool resub = true, const VectorSP &filter = nullptr,
255327
bool msgAsTable = false, bool allowExists = false,
256328
string userName="", string password="",
257-
const StreamDeserializerSP &blobDeserializer = nullptr);
329+
const StreamDeserializerSP &blobDeserializer = nullptr,
330+
const std::vector<std::string> &backupSites = std::vector<std::string>(),
331+
int resubTimeout = 100, bool subOnce = false);
258332
void unsubscribe(string host, int port, string tableName, string actionName = DEFAULT_ACTION_NAME);
259333
};
260334

‎core/include/Types.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ enum class VECTOR_TYPE {ARRAY, BIGARRAY, SUBVECTOR, INDEXVECTOR, REPVECTOR, IOTV
9494
*
9595
* In case we have to change the value, please review the constructor of Constant, Vector, Set, Dictionary, and Table.
9696
*/
97-
enum DATA_FORM {DF_SCALAR,DF_VECTOR,DF_PAIR,DF_MATRIX,DF_SET,DF_DICTIONARY,DF_TABLE,DF_CHART,DF_CHUNK};
97+
enum DATA_FORM {DF_SCALAR,DF_VECTOR,DF_PAIR,DF_MATRIX,DF_SET,DF_DICTIONARY,DF_TABLE,DF_CHART,DF_CHUNK, DF_SYSOBJ, MAX_DATA_FORMS};
9898

9999
enum WORD_TYPE {KEYWORD,ENUM,CONSTANT,VARIABLE,FUNCNAME,GLOBALVARIABLE,OPERATOR,FUNCPATTERN,ASSIGNMENT,REFERENCE,BRACKET,DELIMITOR,COMMA,NAMESPACE,UNKNOWN};
100100

‎core/src/ConstantImp.cpp‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,11 @@ IO_ERR StringVector::deserialize(DataInputStream* in, INDEX indexStart, INDEX ta
3030
auto readBlob = [&](string& value) -> IO_ERR {
3131
IO_ERR ret;
3232
int len;
33-
size_t acLen = 0;
3433
if ((ret = in->readInt(len)) != OK)
3534
return ret;
3635
std::unique_ptr<char[]> buf(new char[len]);
37-
if ((ret = in->readBytes(buf.get(), len, acLen)) != OK)
38-
return ret;
39-
if (acLen != (size_t) len)
40-
return INVALIDDATA;
41-
36+
if ((ret = in->read(buf.get(), len)) != OK)
37+
return ret;
4238
value.clear();
4339
value.append(buf.get(), len);
4440
return ret;

0 commit comments

Comments
 (0)