minor changes
This commit is contained in:
parent
4fe44df5b8
commit
26adff1fa5
|
@ -43,24 +43,22 @@
|
||||||
* An encoding of midnight at the end of the day as 24:00:00 - ie. midnight
|
* An encoding of midnight at the end of the day as 24:00:00 - ie. midnight
|
||||||
* tomorrow - (allowable under ISO 8601) is supported.
|
* tomorrow - (allowable under ISO 8601) is supported.
|
||||||
*/
|
*/
|
||||||
static int64_t user_mktime64(const unsigned int year0, const unsigned int mon0,
|
static int64_t user_mktime64(const uint32_t year0, const uint32_t mon0, const uint32_t day, const uint32_t hour,
|
||||||
const unsigned int day, const unsigned int hour,
|
const uint32_t min, const uint32_t sec, int64_t time_zone) {
|
||||||
const unsigned int min, const unsigned int sec, int64_t time_zone)
|
uint32_t mon = mon0, year = year0;
|
||||||
{
|
|
||||||
unsigned int mon = mon0, year = year0;
|
|
||||||
|
|
||||||
/* 1..12 -> 11,12,1..10 */
|
/* 1..12 -> 11,12,1..10 */
|
||||||
if (0 >= (int) (mon -= 2)) {
|
if (0 >= (int32_t)(mon -= 2)) {
|
||||||
mon += 12; /* Puts Feb last since it has leap day */
|
mon += 12; /* Puts Feb last since it has leap day */
|
||||||
year -= 1;
|
year -= 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
//int64_t res = (((((int64_t) (year/4 - year/100 + year/400 + 367*mon/12 + day) +
|
// int64_t res = (((((int64_t) (year/4 - year/100 + year/400 + 367*mon/12 + day) +
|
||||||
// year*365 - 719499)*24 + hour)*60 + min)*60 + sec);
|
// year*365 - 719499)*24 + hour)*60 + min)*60 + sec);
|
||||||
int64_t res;
|
int64_t res;
|
||||||
res = 367*((int64_t)mon)/12;
|
res = 367 * ((int64_t)mon) / 12;
|
||||||
res += year/4 - year/100 + year/400 + day + ((int64_t)year)*365 - 719499;
|
res += year / 4 - year / 100 + year / 400 + day + ((int64_t)year) * 365 - 719499;
|
||||||
res = res*24;
|
res = res * 24;
|
||||||
res = ((res + hour) * 60 + min) * 60 + sec;
|
res = ((res + hour) * 60 + min) * 60 + sec;
|
||||||
|
|
||||||
return (res + time_zone);
|
return (res + time_zone);
|
||||||
|
@ -71,9 +69,9 @@ static int64_t m_deltaUtc = 0;
|
||||||
void deltaToUtcInitOnce() {
|
void deltaToUtcInitOnce() {
|
||||||
struct tm tm = {0};
|
struct tm tm = {0};
|
||||||
|
|
||||||
(void)strptime("1970-01-01 00:00:00", (const char *)("%Y-%m-%d %H:%M:%S"), &tm);
|
(void)strptime("1970-01-01 00:00:00", (const char*)("%Y-%m-%d %H:%M:%S"), &tm);
|
||||||
m_deltaUtc = (int64_t)mktime(&tm);
|
m_deltaUtc = (int64_t)mktime(&tm);
|
||||||
//printf("====delta:%lld\n\n", seconds);
|
// printf("====delta:%lld\n\n", seconds);
|
||||||
}
|
}
|
||||||
|
|
||||||
static int64_t parseFraction(char* str, char** end, int32_t timePrec);
|
static int64_t parseFraction(char* str, char** end, int32_t timePrec);
|
||||||
|
@ -81,12 +79,10 @@ static int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timeP
|
||||||
static int32_t parseLocaltime(char* timestr, int64_t* time, int32_t timePrec);
|
static int32_t parseLocaltime(char* timestr, int64_t* time, int32_t timePrec);
|
||||||
static int32_t parseLocaltimeDst(char* timestr, int64_t* time, int32_t timePrec);
|
static int32_t parseLocaltimeDst(char* timestr, int64_t* time, int32_t timePrec);
|
||||||
static char* forwardToTimeStringEnd(char* str);
|
static char* forwardToTimeStringEnd(char* str);
|
||||||
static bool checkTzPresent(const char *str, int32_t len);
|
static bool checkTzPresent(const char* str, int32_t len);
|
||||||
|
|
||||||
static int32_t (*parseLocaltimeFp[]) (char* timestr, int64_t* time, int32_t timePrec) = {
|
static int32_t (*parseLocaltimeFp[])(char* timestr, int64_t* time, int32_t timePrec) = {parseLocaltime,
|
||||||
parseLocaltime,
|
parseLocaltimeDst};
|
||||||
parseLocaltimeDst
|
|
||||||
};
|
|
||||||
|
|
||||||
int32_t taosParseTime(const char* timestr, int64_t* time, int32_t len, int32_t timePrec, int8_t day_light) {
|
int32_t taosParseTime(const char* timestr, int64_t* time, int32_t len, int32_t timePrec, int8_t day_light) {
|
||||||
/* parse datatime string in with tz */
|
/* parse datatime string in with tz */
|
||||||
|
@ -103,8 +99,8 @@ bool checkTzPresent(const char* str, int32_t len) {
|
||||||
char* seg = forwardToTimeStringEnd((char*)str);
|
char* seg = forwardToTimeStringEnd((char*)str);
|
||||||
int32_t seg_len = len - (int32_t)(seg - str);
|
int32_t seg_len = len - (int32_t)(seg - str);
|
||||||
|
|
||||||
char *c = &seg[seg_len - 1];
|
char* c = &seg[seg_len - 1];
|
||||||
for (int i = 0; i < seg_len; ++i) {
|
for (int32_t i = 0; i < seg_len; ++i) {
|
||||||
if (*c == 'Z' || *c == 'z' || *c == '+' || *c == '-') {
|
if (*c == 'Z' || *c == 'z' || *c == '+' || *c == '-') {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
@ -199,13 +195,12 @@ int32_t parseTimezone(char* str, int64_t* tzOffset) {
|
||||||
i += 2;
|
i += 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
//return error if there're illegal charaters after min(2 Digits)
|
// return error if there're illegal charaters after min(2 Digits)
|
||||||
char *minStr = &str[i];
|
char* minStr = &str[i];
|
||||||
if (minStr[1] != '\0' && minStr[2] != '\0') {
|
if (minStr[1] != '\0' && minStr[2] != '\0') {
|
||||||
return -1;
|
return -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
int64_t minute = strnatoi(&str[i], 2);
|
int64_t minute = strnatoi(&str[i], 2);
|
||||||
if (minute > 59) {
|
if (minute > 59) {
|
||||||
return -1;
|
return -1;
|
||||||
|
@ -233,9 +228,8 @@ int32_t parseTimezone(char* str, int64_t* tzOffset) {
|
||||||
* 2013-04-12T15:52:01.123+0800
|
* 2013-04-12T15:52:01.123+0800
|
||||||
*/
|
*/
|
||||||
int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timePrec, char delim) {
|
int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timePrec, char delim) {
|
||||||
|
int64_t factor =
|
||||||
int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
|
(timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
||||||
(timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
|
||||||
int64_t tzOffset = 0;
|
int64_t tzOffset = 0;
|
||||||
|
|
||||||
struct tm tm = {0};
|
struct tm tm = {0};
|
||||||
|
@ -255,8 +249,8 @@ int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timePrec, ch
|
||||||
|
|
||||||
/* mktime will be affected by TZ, set by using taos_options */
|
/* mktime will be affected by TZ, set by using taos_options */
|
||||||
#ifdef WINDOWS
|
#ifdef WINDOWS
|
||||||
int64_t seconds = user_mktime64(tm.tm_year+1900, tm.tm_mon+1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, 0);
|
int64_t seconds = user_mktime64(tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, 0);
|
||||||
//int64_t seconds = gmtime(&tm);
|
// int64_t seconds = gmtime(&tm);
|
||||||
#else
|
#else
|
||||||
int64_t seconds = timegm(&tm);
|
int64_t seconds = timegm(&tm);
|
||||||
#endif
|
#endif
|
||||||
|
@ -320,7 +314,8 @@ int32_t parseLocaltime(char* timestr, int64_t* time, int32_t timePrec) {
|
||||||
#endif
|
#endif
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
int64_t seconds = user_mktime64(tm.tm_year+1900, tm.tm_mon+1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, timezone);
|
int64_t seconds =
|
||||||
|
user_mktime64(tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec, timezone);
|
||||||
|
|
||||||
int64_t fraction = 0;
|
int64_t fraction = 0;
|
||||||
|
|
||||||
|
@ -331,8 +326,8 @@ int32_t parseLocaltime(char* timestr, int64_t* time, int32_t timePrec) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
|
int64_t factor =
|
||||||
(timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
(timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
||||||
*time = factor * seconds + fraction;
|
*time = factor * seconds + fraction;
|
||||||
|
|
||||||
return 0;
|
return 0;
|
||||||
|
@ -360,27 +355,22 @@ int32_t parseLocaltimeDst(char* timestr, int64_t* time, int32_t timePrec) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
|
int64_t factor =
|
||||||
(timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
(timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
|
||||||
*time = factor * seconds + fraction;
|
*time = factor * seconds + fraction;
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
|
int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
|
||||||
assert(fromPrecision == TSDB_TIME_PRECISION_MILLI ||
|
assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
|
||||||
fromPrecision == TSDB_TIME_PRECISION_MICRO ||
|
|
||||||
fromPrecision == TSDB_TIME_PRECISION_NANO);
|
fromPrecision == TSDB_TIME_PRECISION_NANO);
|
||||||
assert(toPrecision == TSDB_TIME_PRECISION_MILLI ||
|
assert(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO ||
|
||||||
toPrecision == TSDB_TIME_PRECISION_MICRO ||
|
|
||||||
toPrecision == TSDB_TIME_PRECISION_NANO);
|
toPrecision == TSDB_TIME_PRECISION_NANO);
|
||||||
static double factors[3][3] = { {1., 1000., 1000000.},
|
static double factors[3][3] = {{1., 1000., 1000000.}, {1.0 / 1000, 1., 1000.}, {1.0 / 1000000, 1.0 / 1000, 1.}};
|
||||||
{1.0 / 1000, 1., 1000.},
|
|
||||||
{1.0 / 1000000, 1.0 / 1000, 1.} };
|
|
||||||
return (int64_t)((double)time * factors[fromPrecision][toPrecision]);
|
return (int64_t)((double)time * factors[fromPrecision][toPrecision]);
|
||||||
}
|
}
|
||||||
|
|
||||||
static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t timePrecision) {
|
static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t timePrecision) {
|
||||||
|
|
||||||
switch (unit) {
|
switch (unit) {
|
||||||
case 's':
|
case 's':
|
||||||
(*result) = convertTimePrecision(val * MILLISECOND_PER_SECOND, TSDB_TIME_PRECISION_MILLI, timePrecision);
|
(*result) = convertTimePrecision(val * MILLISECOND_PER_SECOND, TSDB_TIME_PRECISION_MILLI, timePrecision);
|
||||||
|
@ -427,7 +417,8 @@ static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t time
|
||||||
* d - Days (24 hours)
|
* d - Days (24 hours)
|
||||||
* w - Weeks (7 days)
|
* w - Weeks (7 days)
|
||||||
*/
|
*/
|
||||||
int32_t parseAbsoluteDuration(const char* token, int32_t tokenlen, int64_t* duration, char* unit, int32_t timePrecision) {
|
int32_t parseAbsoluteDuration(const char* token, int32_t tokenlen, int64_t* duration, char* unit,
|
||||||
|
int32_t timePrecision) {
|
||||||
errno = 0;
|
errno = 0;
|
||||||
char* endPtr = NULL;
|
char* endPtr = NULL;
|
||||||
|
|
||||||
|
@ -476,7 +467,7 @@ int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision) {
|
||||||
struct tm tm;
|
struct tm tm;
|
||||||
time_t tt = (time_t)(t / TSDB_TICK_PER_SECOND(precision));
|
time_t tt = (time_t)(t / TSDB_TICK_PER_SECOND(precision));
|
||||||
localtime_r(&tt, &tm);
|
localtime_r(&tt, &tm);
|
||||||
int mon = tm.tm_year * 12 + tm.tm_mon + (int)duration;
|
int32_t mon = tm.tm_year * 12 + tm.tm_mon + (int32_t)duration;
|
||||||
tm.tm_year = mon / 12;
|
tm.tm_year = mon / 12;
|
||||||
tm.tm_mon = mon % 12;
|
tm.tm_mon = mon % 12;
|
||||||
|
|
||||||
|
@ -499,11 +490,11 @@ int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char
|
||||||
struct tm tm;
|
struct tm tm;
|
||||||
time_t t = (time_t)skey;
|
time_t t = (time_t)skey;
|
||||||
localtime_r(&t, &tm);
|
localtime_r(&t, &tm);
|
||||||
int smon = tm.tm_year * 12 + tm.tm_mon;
|
int32_t smon = tm.tm_year * 12 + tm.tm_mon;
|
||||||
|
|
||||||
t = (time_t)ekey;
|
t = (time_t)ekey;
|
||||||
localtime_r(&t, &tm);
|
localtime_r(&t, &tm);
|
||||||
int emon = tm.tm_year * 12 + tm.tm_mon;
|
int32_t emon = tm.tm_year * 12 + tm.tm_mon;
|
||||||
|
|
||||||
if (unit == 'y') {
|
if (unit == 'y') {
|
||||||
interval *= 12;
|
interval *= 12;
|
||||||
|
@ -531,10 +522,10 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
|
||||||
|
|
||||||
if (pInterval->slidingUnit == 'y') {
|
if (pInterval->slidingUnit == 'y') {
|
||||||
tm.tm_mon = 0;
|
tm.tm_mon = 0;
|
||||||
tm.tm_year = (int)(tm.tm_year / pInterval->sliding * pInterval->sliding);
|
tm.tm_year = (int32_t)(tm.tm_year / pInterval->sliding * pInterval->sliding);
|
||||||
} else {
|
} else {
|
||||||
int mon = tm.tm_year * 12 + tm.tm_mon;
|
int32_t mon = tm.tm_year * 12 + tm.tm_mon;
|
||||||
mon = (int)(mon / pInterval->sliding * pInterval->sliding);
|
mon = (int32_t)(mon / pInterval->sliding * pInterval->sliding);
|
||||||
tm.tm_year = mon / 12;
|
tm.tm_year = mon / 12;
|
||||||
tm.tm_mon = mon % 12;
|
tm.tm_mon = mon % 12;
|
||||||
}
|
}
|
||||||
|
@ -552,12 +543,12 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
|
||||||
* but in case of DST, the start time of one day need to be dynamically decided.
|
* but in case of DST, the start time of one day need to be dynamically decided.
|
||||||
*/
|
*/
|
||||||
// todo refactor to extract function that is available for Linux/Windows/Mac platform
|
// todo refactor to extract function that is available for Linux/Windows/Mac platform
|
||||||
#if defined(WINDOWS) && _MSC_VER >= 1900
|
#if defined(WINDOWS) && _MSC_VER >= 1900
|
||||||
// see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
|
// see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
|
||||||
int64_t timezone = _timezone;
|
int64_t timezone = _timezone;
|
||||||
int32_t daylight = _daylight;
|
int32_t daylight = _daylight;
|
||||||
char** tzname = _tzname;
|
char** tzname = _tzname;
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
start += (int64_t)(timezone * TSDB_TICK_PER_SECOND(precision));
|
start += (int64_t)(timezone * TSDB_TICK_PER_SECOND(precision));
|
||||||
}
|
}
|
||||||
|
@ -568,7 +559,7 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
|
||||||
if (start < 0 || INT64_MAX - start > pInterval->interval - 1) {
|
if (start < 0 || INT64_MAX - start > pInterval->interval - 1) {
|
||||||
end = start + pInterval->interval - 1;
|
end = start + pInterval->interval - 1;
|
||||||
|
|
||||||
while(end < t && ((start + pInterval->sliding) <= INT64_MAX)) { // move forward to the correct time window
|
while (end < t && ((start + pInterval->sliding) <= INT64_MAX)) { // move forward to the correct time window
|
||||||
start += pInterval->sliding;
|
start += pInterval->sliding;
|
||||||
|
|
||||||
if (start < 0 || INT64_MAX - start > pInterval->interval - 1) {
|
if (start < 0 || INT64_MAX - start > pInterval->interval - 1) {
|
||||||
|
@ -619,7 +610,7 @@ const char* fmtts(int64_t ts) {
|
||||||
buf[pos++] = ' ';
|
buf[pos++] = ' ';
|
||||||
}
|
}
|
||||||
pos += strftime(buf + pos, sizeof(buf), "ms=%Y-%m-%d %H:%M:%S", &tm);
|
pos += strftime(buf + pos, sizeof(buf), "ms=%Y-%m-%d %H:%M:%S", &tm);
|
||||||
pos += sprintf(buf + pos, ".%03d", (int)(ts % 1000));
|
pos += sprintf(buf + pos, ".%03d", (int32_t)(ts % 1000));
|
||||||
}
|
}
|
||||||
|
|
||||||
{
|
{
|
||||||
|
@ -631,7 +622,7 @@ const char* fmtts(int64_t ts) {
|
||||||
buf[pos++] = ' ';
|
buf[pos++] = ' ';
|
||||||
}
|
}
|
||||||
pos += strftime(buf + pos, sizeof(buf), "us=%Y-%m-%d %H:%M:%S", &tm);
|
pos += strftime(buf + pos, sizeof(buf), "us=%Y-%m-%d %H:%M:%S", &tm);
|
||||||
pos += sprintf(buf + pos, ".%06d", (int)(ts % 1000000));
|
pos += sprintf(buf + pos, ".%06d", (int32_t)(ts % 1000000));
|
||||||
}
|
}
|
||||||
|
|
||||||
return buf;
|
return buf;
|
||||||
|
|
|
@ -115,7 +115,7 @@ STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete) {
|
||||||
// check the ts order
|
// check the ts order
|
||||||
pTSBuf->tsOrder = header.tsOrder;
|
pTSBuf->tsOrder = header.tsOrder;
|
||||||
if (pTSBuf->tsOrder != TSDB_ORDER_ASC && pTSBuf->tsOrder != TSDB_ORDER_DESC) {
|
if (pTSBuf->tsOrder != TSDB_ORDER_ASC && pTSBuf->tsOrder != TSDB_ORDER_DESC) {
|
||||||
// tscError("invalid order info in buf:%d", pTSBuf->tsOrder);
|
// tscError("invalid order info in buf:%d", pTSBuf->tsOrder);
|
||||||
tsBufDestroy(pTSBuf);
|
tsBufDestroy(pTSBuf);
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
@ -128,7 +128,7 @@ STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete) {
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
//int64_t pos = ftell(pTSBuf->pFile); //pos not used
|
// int64_t pos = ftell(pTSBuf->pFile); //pos not used
|
||||||
sz = taosReadFile(pTSBuf->pFile, buf, infoSize);
|
sz = taosReadFile(pTSBuf->pFile, buf, infoSize);
|
||||||
UNUSED(sz);
|
UNUSED(sz);
|
||||||
|
|
||||||
|
@ -154,8 +154,9 @@ STSBuf* tsBufCreateFromFile(const char* path, bool autoDelete) {
|
||||||
// ascending by default
|
// ascending by default
|
||||||
pTSBuf->cur.order = TSDB_ORDER_ASC;
|
pTSBuf->cur.order = TSDB_ORDER_ASC;
|
||||||
|
|
||||||
// tscDebug("create tsBuf from file:%s, fd:%d, size:%d, numOfGroups:%d, autoDelete:%d", pTSBuf->path, fileno(pTSBuf->pFile),
|
// tscDebug("create tsBuf from file:%s, fd:%d, size:%d, numOfGroups:%d, autoDelete:%d", pTSBuf->path,
|
||||||
// pTSBuf->fileSize, pTSBuf->numOfGroups, pTSBuf->autoDelete);
|
// fileno(pTSBuf->pFile),
|
||||||
|
// pTSBuf->fileSize, pTSBuf->numOfGroups, pTSBuf->autoDelete);
|
||||||
|
|
||||||
return pTSBuf;
|
return pTSBuf;
|
||||||
}
|
}
|
||||||
|
@ -176,10 +177,10 @@ void* tsBufDestroy(STSBuf* pTSBuf) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if (pTSBuf->autoDelete) {
|
if (pTSBuf->autoDelete) {
|
||||||
// ("tsBuf %p destroyed, delete tmp file:%s", pTSBuf, pTSBuf->path);
|
// ("tsBuf %p destroyed, delete tmp file:%s", pTSBuf, pTSBuf->path);
|
||||||
remove(pTSBuf->path);
|
remove(pTSBuf->path);
|
||||||
} else {
|
} else {
|
||||||
// tscDebug("tsBuf %p destroyed, tmp file:%s, remains", pTSBuf, pTSBuf->path);
|
// tscDebug("tsBuf %p destroyed, tmp file:%s, remains", pTSBuf, pTSBuf->path);
|
||||||
}
|
}
|
||||||
|
|
||||||
taosVariantDestroy(&pTSBuf->block.tag);
|
taosVariantDestroy(&pTSBuf->block.tag);
|
||||||
|
@ -240,7 +241,7 @@ static void shrinkBuffer(STSList* ptsData) {
|
||||||
// shrink tmp buffer size if it consumes too many memory compared to the pre-defined size
|
// shrink tmp buffer size if it consumes too many memory compared to the pre-defined size
|
||||||
if (ptsData->allocSize >= ptsData->threshold * 2) {
|
if (ptsData->allocSize >= ptsData->threshold * 2) {
|
||||||
char* rawBuf = realloc(ptsData->rawBuf, MEM_BUF_SIZE);
|
char* rawBuf = realloc(ptsData->rawBuf, MEM_BUF_SIZE);
|
||||||
if(rawBuf) {
|
if (rawBuf) {
|
||||||
ptsData->rawBuf = rawBuf;
|
ptsData->rawBuf = rawBuf;
|
||||||
ptsData->allocSize = MEM_BUF_SIZE;
|
ptsData->allocSize = MEM_BUF_SIZE;
|
||||||
}
|
}
|
||||||
|
@ -265,9 +266,8 @@ static void writeDataToDisk(STSBuf* pTSBuf) {
|
||||||
STSList* pTsData = &pTSBuf->tsData;
|
STSList* pTsData = &pTSBuf->tsData;
|
||||||
|
|
||||||
pBlock->numOfElem = pTsData->len / TSDB_KEYSIZE;
|
pBlock->numOfElem = pTsData->len / TSDB_KEYSIZE;
|
||||||
pBlock->compLen =
|
pBlock->compLen = tsCompressTimestamp(pTsData->rawBuf, pTsData->len, pTsData->len / TSDB_KEYSIZE, pBlock->payload,
|
||||||
tsCompressTimestamp(pTsData->rawBuf, pTsData->len, pTsData->len/TSDB_KEYSIZE, pBlock->payload, pTsData->allocSize,
|
pTsData->allocSize, TWO_STAGE_COMP, pTSBuf->assistBuf, pTSBuf->bufSize);
|
||||||
TWO_STAGE_COMP, pTSBuf->assistBuf, pTSBuf->bufSize);
|
|
||||||
|
|
||||||
int64_t r = taosLSeekFile(pTSBuf->pFile, pTSBuf->fileSize, SEEK_SET);
|
int64_t r = taosLSeekFile(pTSBuf->pFile, pTSBuf->fileSize, SEEK_SET);
|
||||||
assert(r == 0);
|
assert(r == 0);
|
||||||
|
@ -289,10 +289,10 @@ static void writeDataToDisk(STSBuf* pTSBuf) {
|
||||||
} else if (pBlock->tag.nType == TSDB_DATA_TYPE_FLOAT) {
|
} else if (pBlock->tag.nType == TSDB_DATA_TYPE_FLOAT) {
|
||||||
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
||||||
float tfloat = (float)pBlock->tag.d;
|
float tfloat = (float)pBlock->tag.d;
|
||||||
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &tfloat, (size_t) pBlock->tag.nLen);
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &tfloat, (size_t)pBlock->tag.nLen);
|
||||||
} else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) {
|
} else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) {
|
||||||
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
||||||
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.i, (size_t) pBlock->tag.nLen);
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &pBlock->tag.i, (size_t)pBlock->tag.nLen);
|
||||||
} else {
|
} else {
|
||||||
trueLen = 0;
|
trueLen = 0;
|
||||||
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &trueLen, sizeof(pBlock->tag.nLen));
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &trueLen, sizeof(pBlock->tag.nLen));
|
||||||
|
@ -303,7 +303,7 @@ static void writeDataToDisk(STSBuf* pTSBuf) {
|
||||||
taosWriteFile(pTSBuf->pFile, pBlock->payload, (size_t)pBlock->compLen);
|
taosWriteFile(pTSBuf->pFile, pBlock->payload, (size_t)pBlock->compLen);
|
||||||
taosWriteFile(pTSBuf->pFile, &pBlock->compLen, sizeof(pBlock->compLen));
|
taosWriteFile(pTSBuf->pFile, &pBlock->compLen, sizeof(pBlock->compLen));
|
||||||
|
|
||||||
metaLen += (int32_t) taosWriteFile(pTSBuf->pFile, &trueLen, sizeof(pBlock->tag.nLen));
|
metaLen += (int32_t)taosWriteFile(pTSBuf->pFile, &trueLen, sizeof(pBlock->tag.nLen));
|
||||||
assert(metaLen == getTagAreaLength(&pBlock->tag));
|
assert(metaLen == getTagAreaLength(&pBlock->tag));
|
||||||
|
|
||||||
int32_t blockSize = metaLen + sizeof(pBlock->numOfElem) + sizeof(pBlock->compLen) * 2 + pBlock->compLen;
|
int32_t blockSize = metaLen + sizeof(pBlock->numOfElem) + sizeof(pBlock->compLen) * 2 + pBlock->compLen;
|
||||||
|
@ -347,7 +347,7 @@ STSBlock* readDataFromDisk(STSBuf* pTSBuf, int32_t order, bool decomp) {
|
||||||
* set the right position for the reversed traverse, the reversed traverse is started from
|
* set the right position for the reversed traverse, the reversed traverse is started from
|
||||||
* the end of each comp data block
|
* the end of each comp data block
|
||||||
*/
|
*/
|
||||||
int32_t prev = -(int32_t) (sizeof(pBlock->padding) + sizeof(pBlock->tag.nLen));
|
int32_t prev = -(int32_t)(sizeof(pBlock->padding) + sizeof(pBlock->tag.nLen));
|
||||||
int32_t ret = taosLSeekFile(pTSBuf->pFile, prev, SEEK_CUR);
|
int32_t ret = taosLSeekFile(pTSBuf->pFile, prev, SEEK_CUR);
|
||||||
size_t sz = taosReadFile(pTSBuf->pFile, &pBlock->padding, sizeof(pBlock->padding));
|
size_t sz = taosReadFile(pTSBuf->pFile, &pBlock->padding, sizeof(pBlock->padding));
|
||||||
sz = taosReadFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
sz = taosReadFile(pTSBuf->pFile, &pBlock->tag.nLen, sizeof(pBlock->tag.nLen));
|
||||||
|
@ -376,11 +376,11 @@ STSBlock* readDataFromDisk(STSBuf* pTSBuf, int32_t order, bool decomp) {
|
||||||
UNUSED(sz);
|
UNUSED(sz);
|
||||||
} else if (pBlock->tag.nType == TSDB_DATA_TYPE_FLOAT) {
|
} else if (pBlock->tag.nType == TSDB_DATA_TYPE_FLOAT) {
|
||||||
float tfloat = 0;
|
float tfloat = 0;
|
||||||
sz = taosReadFile(pTSBuf->pFile, &tfloat, (size_t) pBlock->tag.nLen);
|
sz = taosReadFile(pTSBuf->pFile, &tfloat, (size_t)pBlock->tag.nLen);
|
||||||
pBlock->tag.d = (double)tfloat;
|
pBlock->tag.d = (double)tfloat;
|
||||||
UNUSED(sz);
|
UNUSED(sz);
|
||||||
} else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) { //TODO check the return value
|
} else if (pBlock->tag.nType != TSDB_DATA_TYPE_NULL) { // TODO check the return value
|
||||||
sz = taosReadFile(pTSBuf->pFile, &pBlock->tag.i, (size_t) pBlock->tag.nLen);
|
sz = taosReadFile(pTSBuf->pFile, &pBlock->tag.i, (size_t)pBlock->tag.nLen);
|
||||||
UNUSED(sz);
|
UNUSED(sz);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -574,7 +574,7 @@ static int32_t tsBufFindBlockByTag(STSBuf* pTSBuf, STSGroupBlockInfo* pBlockInfo
|
||||||
}
|
}
|
||||||
|
|
||||||
if (taosVariantCompare(&pTSBuf->block.tag, tag) == 0) {
|
if (taosVariantCompare(&pTSBuf->block.tag, tag) == 0) {
|
||||||
return (pTSBuf->cur.order == TSDB_ORDER_ASC)? i: (pBlockInfo->numOfBlocks - (i + 1));
|
return (pTSBuf->cur.order == TSDB_ORDER_ASC) ? i : (pBlockInfo->numOfBlocks - (i + 1));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -660,13 +660,14 @@ int32_t STSBufUpdateHeader(STSBuf* pTSBuf, STSBufFileHeader* pHeader) {
|
||||||
|
|
||||||
int32_t r = taosLSeekFile(pTSBuf->pFile, 0, SEEK_SET);
|
int32_t r = taosLSeekFile(pTSBuf->pFile, 0, SEEK_SET);
|
||||||
if (r != 0) {
|
if (r != 0) {
|
||||||
// qError("fseek failed, errno:%d", errno);
|
// qError("fseek failed, errno:%d", errno);
|
||||||
return -1;
|
return -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t ws = taosWriteFile(pTSBuf->pFile, pHeader, sizeof(STSBufFileHeader));
|
size_t ws = taosWriteFile(pTSBuf->pFile, pHeader, sizeof(STSBufFileHeader));
|
||||||
if (ws != 1) {
|
if (ws != 1) {
|
||||||
// qError("ts update header fwrite failed, size:%d, expected size:%d", (int32_t)ws, (int32_t)sizeof(STSBufFileHeader));
|
// qError("ts update header fwrite failed, size:%d, expected size:%d", (int32_t)ws,
|
||||||
|
// (int32_t)sizeof(STSBufFileHeader));
|
||||||
return -1;
|
return -1;
|
||||||
}
|
}
|
||||||
return 0;
|
return 0;
|
||||||
|
@ -803,7 +804,7 @@ int32_t tsBufMerge(STSBuf* pDestBuf, const STSBuf* pSrcBuf) {
|
||||||
tsBufFlush(pDestBuf);
|
tsBufFlush(pDestBuf);
|
||||||
|
|
||||||
// compared with the last vnode id
|
// compared with the last vnode id
|
||||||
int32_t id = tsBufGetLastGroupInfo((STSBuf*) pSrcBuf)->info.id;
|
int32_t id = tsBufGetLastGroupInfo((STSBuf*)pSrcBuf)->info.id;
|
||||||
if (id != tsBufGetLastGroupInfo(pDestBuf)->info.id) {
|
if (id != tsBufGetLastGroupInfo(pDestBuf)->info.id) {
|
||||||
int32_t oldSize = pDestBuf->numOfGroups;
|
int32_t oldSize = pDestBuf->numOfGroups;
|
||||||
int32_t newSize = oldSize + pSrcBuf->numOfGroups;
|
int32_t newSize = oldSize + pSrcBuf->numOfGroups;
|
||||||
|
@ -887,13 +888,13 @@ STSBuf* tsBufCreateFromCompBlocks(const char* pData, int32_t numOfBlocks, int32_
|
||||||
|
|
||||||
int32_t ret = taosLSeekFile(pTSBuf->pFile, pBlockInfo->offset, SEEK_SET);
|
int32_t ret = taosLSeekFile(pTSBuf->pFile, pBlockInfo->offset, SEEK_SET);
|
||||||
if (ret == -1) {
|
if (ret == -1) {
|
||||||
// qError("fseek failed, errno:%d", errno);
|
// qError("fseek failed, errno:%d", errno);
|
||||||
tsBufDestroy(pTSBuf);
|
tsBufDestroy(pTSBuf);
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
size_t sz = taosWriteFile(pTSBuf->pFile, (void*)pData, len);
|
size_t sz = taosWriteFile(pTSBuf->pFile, (void*)pData, len);
|
||||||
if (sz != len) {
|
if (sz != len) {
|
||||||
// qError("ts data fwrite failed, write size:%d, expected size:%d", (int32_t)sz, len);
|
// qError("ts data fwrite failed, write size:%d, expected size:%d", (int32_t)sz, len);
|
||||||
tsBufDestroy(pTSBuf);
|
tsBufDestroy(pTSBuf);
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
@ -910,11 +911,11 @@ STSBuf* tsBufCreateFromCompBlocks(const char* pData, int32_t numOfBlocks, int32_
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO taosFsync??
|
// TODO taosFsync??
|
||||||
// if (taosFsync(fileno(pTSBuf->pFile)) == -1) {
|
// if (taosFsync(fileno(pTSBuf->pFile)) == -1) {
|
||||||
//// qError("fsync failed, errno:%d", errno);
|
//// qError("fsync failed, errno:%d", errno);
|
||||||
// tsBufDestroy(pTSBuf);
|
// tsBufDestroy(pTSBuf);
|
||||||
// return NULL;
|
// return NULL;
|
||||||
// }
|
// }
|
||||||
|
|
||||||
return pTSBuf;
|
return pTSBuf;
|
||||||
}
|
}
|
||||||
|
@ -1076,28 +1077,28 @@ void tsBufGetGroupIdList(STSBuf* pTSBuf, int32_t* num, int32_t** id) {
|
||||||
|
|
||||||
(*id) = malloc(tsBufGetNumOfGroup(pTSBuf) * sizeof(int32_t));
|
(*id) = malloc(tsBufGetNumOfGroup(pTSBuf) * sizeof(int32_t));
|
||||||
|
|
||||||
for(int32_t i = 0; i < size; ++i) {
|
for (int32_t i = 0; i < size; ++i) {
|
||||||
(*id)[i] = pTSBuf->pData[i].info.id;
|
(*id)[i] = pTSBuf->pData[i].info.id;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t dumpFileBlockByGroupId(STSBuf* pTSBuf, int32_t groupIndex, void* buf, int32_t* len, int32_t* numOfBlocks) {
|
int32_t dumpFileBlockByGroupId(STSBuf* pTSBuf, int32_t groupIndex, void* buf, int32_t* len, int32_t* numOfBlocks) {
|
||||||
assert(groupIndex >= 0 && groupIndex < pTSBuf->numOfGroups);
|
assert(groupIndex >= 0 && groupIndex < pTSBuf->numOfGroups);
|
||||||
STSGroupBlockInfo *pBlockInfo = &pTSBuf->pData[groupIndex].info;
|
STSGroupBlockInfo* pBlockInfo = &pTSBuf->pData[groupIndex].info;
|
||||||
|
|
||||||
*len = 0;
|
*len = 0;
|
||||||
*numOfBlocks = 0;
|
*numOfBlocks = 0;
|
||||||
|
|
||||||
if (taosLSeekFile(pTSBuf->pFile, pBlockInfo->offset, SEEK_SET) != 0) {
|
if (taosLSeekFile(pTSBuf->pFile, pBlockInfo->offset, SEEK_SET) != 0) {
|
||||||
int code = TAOS_SYSTEM_ERROR(taosEOFFile(pTSBuf->pFile));
|
int32_t code = TAOS_SYSTEM_ERROR(taosEOFFile(pTSBuf->pFile));
|
||||||
// qError("%p: fseek failed: %s", pSql, tstrerror(code));
|
// qError("%p: fseek failed: %s", pSql, tstrerror(code));
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t s = taosReadFile(pTSBuf->pFile, buf, pBlockInfo->compLen);
|
size_t s = taosReadFile(pTSBuf->pFile, buf, pBlockInfo->compLen);
|
||||||
if (s != pBlockInfo->compLen) {
|
if (s != pBlockInfo->compLen) {
|
||||||
int code = TAOS_SYSTEM_ERROR(taosEOFFile(pTSBuf->pFile));
|
int32_t code = TAOS_SYSTEM_ERROR(taosEOFFile(pTSBuf->pFile));
|
||||||
// tscError("%p: fread didn't return expected data: %s", pSql, tstrerror(code));
|
// tscError("%p: fread didn't return expected data: %s", pSql, tstrerror(code));
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -1120,6 +1121,4 @@ STSElem tsBufFindElemStartPosByTag(STSBuf* pTSBuf, SVariant* pTag) {
|
||||||
return el;
|
return el;
|
||||||
}
|
}
|
||||||
|
|
||||||
bool tsBufIsValidElem(STSElem* pElem) {
|
bool tsBufIsValidElem(STSElem* pElem) { return pElem->id >= 0; }
|
||||||
return pElem->id >= 0;
|
|
||||||
}
|
|
||||||
|
|
Loading…
Reference in New Issue