Skip to content

Commit

Permalink
Big datetime in Datashard TTL (#5664)
Browse files Browse the repository at this point in the history
  • Loading branch information
@azevaykin
azevaykin authored Jun 19, 2024
1 parent c6a6871 commit ba35e13
Show file tree
Hide file tree
Showing 8 changed files with 195 additions and 8 deletions.
3 changes: 3 additions & 0 deletions ydb/core/tx/datashard/datashard__conditional_erase_rows.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -470,6 +470,9 @@ static bool CheckUnit(NScheme::TTypeInfo type, NKikimrSchemeOp::TTTLSettings::EU
case NScheme::NTypeIds::Date:
case NScheme::NTypeIds::Datetime:
case NScheme::NTypeIds::Timestamp:
case NScheme::NTypeIds::Date32:
case NScheme::NTypeIds::Datetime64:
case NScheme::NTypeIds::Timestamp64:
if (unit == NKikimrSchemeOp::TTTLSettings::UNIT_AUTO) {
return true;
} else {
Expand Down
52 changes: 50 additions & 2 deletions ydb/core/tx/datashard/datashard_ut_erase_rows.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -428,14 +428,18 @@ Y_UNIT_TEST_SUITE(EraseRowsTests) {
TProto::TEvEraseResponse::SCHEME_ERROR, "Cell count doesn't match row scheme");
}

void ConditionalEraseShouldSuccess(const TString& ttlColType, EUnit unit, const TString& toUpload, const TString& afterErase) {
void ConditionalEraseShouldSuccess(const TString& ttlColType, EUnit unit, const TString& toUpload, const TString& afterErase, const bool enableDatetime64 = false) {
using TEvResponse = TEvDataShard::TEvConditionalEraseRowsResponse;

NKikimrConfig::TFeatureFlags featureFlags;
featureFlags.SetEnableTableDatetime64(enableDatetime64);

TPortManager pm;
TServerSettings serverSettings(pm.GetPort(2134));
serverSettings
.SetDomainName("Root")
.SetUseRealThreads(false);
.SetUseRealThreads(false)
.SetFeatureFlags(featureFlags);

TServer::TPtr server = new TServer(serverSettings);
auto& runtime = *server->GetRuntime();
Expand Down Expand Up @@ -615,6 +619,50 @@ key = 4, value = (empty maybe)
)");
}

Y_UNIT_TEST(ConditionalEraseRowsShouldEraseOnDate32) {
ConditionalEraseShouldSuccess("Date32", TUnit::AUTO, R"(
UPSERT INTO `/Root/table-1` (key, value) VALUES
(1, CAST("1960-01-01" AS Date32)),
(2, CAST("1970-01-01" AS Date32)),
(3, CAST("1990-03-01" AS Date32)),
(4, CAST("2030-04-15" AS Date32)),
(5, NULL);
)", R"(
key = 4, value = 22019
key = 5, value = (empty maybe)
)", true);
}

Y_UNIT_TEST(ConditionalEraseRowsShouldEraseOnDatetime64) {
ConditionalEraseShouldSuccess("Datetime64", TUnit::AUTO, R"(
UPSERT INTO `/Root/table-1` (key, value) VALUES
(1, CAST("1960-01-01T00:00:00Z" AS Datetime64)),
(2, CAST("1970-01-01T00:00:00Z" AS Datetime64)),
(3, CAST("1990-03-01T00:00:00Z" AS Datetime64)),
(4, CAST("2030-04-15T00:00:00Z" AS Datetime64)),
(5, NULL);
)", R"(
key = 4, value = 1902441600
key = 5, value = (empty maybe)
)", true);
}

Y_UNIT_TEST(ConditionalEraseRowsShouldEraseOnTimestamp64) {
ConditionalEraseShouldSuccess("Timestamp64", TUnit::AUTO, R"(
UPSERT INTO `/Root/table-1` (key, value) VALUES
(1, CAST("1960-01-01T00:00:00.000000Z" AS Timestamp64)),
(2, CAST("1970-01-01T00:00:00.000000Z" AS Timestamp64)),
(3, CAST("1990-03-01T00:00:00.000000Z" AS Timestamp64)),
(4, CAST("2030-04-15T00:00:00.000000Z" AS Timestamp64)),
(5, NULL);
)", R"(
key = 4, value = 1902441600000000
key = 5, value = (empty maybe)
)", true);
}



Y_UNIT_TEST(ConditionalEraseRowsShouldFailOnVariousErrors) {
using TEvResponse = TEvDataShard::TEvConditionalEraseRowsResponse;

Expand Down
36 changes: 30 additions & 6 deletions ydb/core/tx/datashard/erase_rows_condition.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ class TExpirationCondition: public IEraseRowsCondition {
return WallClockDyNumber;
}

bool Check(ui64 value) const {
bool CheckUi64(ui64 value) const {
switch (Type) {
// 'date-type column' mode
case NScheme::NTypeIds::Date:
Expand Down Expand Up @@ -87,7 +87,26 @@ class TExpirationCondition: public IEraseRowsCondition {
}
}

bool Check(TStringBuf value) const {
bool CheckI64(i64 value) const {

// Dates before 1970 are deleted by TTL
if (value < 0)
return true;

switch (Type) {
// 'big date-type column' mode
case NScheme::NTypeIds::Date32:
return TInstant::Days(value) <= WallClockInstant;
case NScheme::NTypeIds::Datetime64:
return TInstant::Seconds(value) <= WallClockInstant;
case NScheme::NTypeIds::Timestamp64:
return TInstant::MicroSeconds(value) <= WallClockInstant;
default:
Y_ABORT("Unreachable");
}
}

bool CheckStr(TStringBuf value) const {
switch (Type) {
// 'value since epoch' mode
case NScheme::NTypeIds::DyNumber:
Expand Down Expand Up @@ -147,15 +166,20 @@ class TExpirationCondition: public IEraseRowsCondition {

switch (Type) {
case NScheme::NTypeIds::Date:
return Check(cell.AsValue<ui16>());
return CheckUi64(cell.AsValue<ui16>());
case NScheme::NTypeIds::Datetime:
case NScheme::NTypeIds::Uint32:
return Check(cell.AsValue<ui32>());
return CheckUi64(cell.AsValue<ui32>());
case NScheme::NTypeIds::Timestamp:
case NScheme::NTypeIds::Uint64:
return Check(cell.AsValue<ui64>());
return CheckUi64(cell.AsValue<ui64>());
case NScheme::NTypeIds::Date32:
return CheckI64(cell.AsValue<i32>());
case NScheme::NTypeIds::Datetime64:
case NScheme::NTypeIds::Timestamp64:
return CheckI64(cell.AsValue<i64>());
case NScheme::NTypeIds::DyNumber:
return Check(cell.AsBuf());
return CheckStr(cell.AsBuf());
default:
return false;
}
Expand Down
16 changes: 16 additions & 0 deletions ydb/core/tx/datashard/read_table_scan.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,22 @@ Y_FORCE_INLINE bool AddCell(TOutValue& row, NScheme::TTypeInfo type, const TCell
val.set_int64_value(value);
break;
}
case NUdf::TDataType<NUdf::TDate32>::Id: {
i32 value;
if (!cell.ToValue(value, err))
return false;
val.set_int32_value(value);
break;
}
case NUdf::TDataType<NUdf::TDatetime64>::Id:
case NUdf::TDataType<NUdf::TTimestamp64>::Id:
case NUdf::TDataType<NUdf::TInterval64>::Id: {
i64 value;
if (!cell.ToValue(value, err))
return false;
val.set_int64_value(value);
break;
}
case NUdf::TDataType<NUdf::TJsonDocument>::Id: {
const auto json = NBinaryJson::SerializeToJson(TStringBuf(cell.Data(), cell.Size()));
val.set_text_value(json);
Expand Down
3 changes: 3 additions & 0 deletions ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2357,6 +2357,9 @@ namespace {
PRINT_PRIMITIVE(Date);
PRINT_PRIMITIVE(Datetime);
PRINT_PRIMITIVE(Timestamp);
PRINT_PRIMITIVE(Date32);
PRINT_PRIMITIVE(Datetime64);
PRINT_PRIMITIVE(Timestamp64);
PRINT_PRIMITIVE(String);
PRINT_PRIMITIVE(DyNumber);

Expand Down
6 changes: 6 additions & 0 deletions ydb/core/tx/schemeshard/schemeshard_validate_ttl.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,12 @@ bool ValidateTtlSettings(const NKikimrSchemeOp::TTTLSettings& ttl,
return false;
}

const TInstant now = TInstant::Now();
if (enabled.GetExpireAfterSeconds() > now.Seconds()) {
errStr = Sprintf("TTL should be less than %" PRIu64 " seconds (%" PRIu64 " days, %" PRIu64 " years). The ttl behaviour is undefined before 1970.", now.Seconds(), now.Days(), now.Days() / 365);
return false;
}

if (enabled.HasSysSettings()) {
const auto& sys = enabled.GetSysSettings();
if (TDuration::FromValue(sys.GetRunInterval()) < subDomain.GetTtlMinRunInterval()) {
Expand Down
23 changes: 23 additions & 0 deletions ydb/core/tx/schemeshard/ut_ttl/ut_ttl.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,29 @@ Y_UNIT_TEST_SUITE(TSchemeShardTTLTests) {
)", {NKikimrScheme::StatusSchemeError});
}

Y_UNIT_TEST(CreateTableShouldFailOnBeforeEpochTTL) {
TTestBasicRuntime runtime;
TTestEnv env(runtime);
ui64 txId = 100;

// An attempt to create 100-year TTL.
// The TTL behaviour is undefined before 1970,
// so it's forbidden.

TestCreateTable(runtime, ++txId, "/MyRoot", R"(
Name: "TTLEnabledTable"
Columns { Name: "key" Type: "Uint64" }
Columns { Name: "modified_at" Type: "Timestamp" }
KeyColumnNames: ["key"]
TTLSettings {
Enabled {
ColumnName: "modified_at"
ExpireAfterSeconds: 3153600000
}
}
)", {NKikimrScheme::StatusSchemeError});
}

void CreateTableOnIndexedTable(NKikimrSchemeOp::EIndexType indexType) {
TTestBasicRuntime runtime;
TTestEnv env(runtime);
Expand Down
64 changes: 64 additions & 0 deletions ydb/core/ydb_convert/ydb_convert.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -307,6 +307,14 @@ Y_FORCE_INLINE void ConvertData(NUdf::TDataTypeId typeId, const NKikimrMiniKQL::
case NUdf::TDataType<NUdf::TInterval>::Id:
res.set_int64_value(value.GetInt64());
break;
case NUdf::TDataType<NUdf::TDate32>::Id:
res.set_int32_value(value.GetInt32());
break;
case NUdf::TDataType<NUdf::TDatetime64>::Id:
case NUdf::TDataType<NUdf::TTimestamp64>::Id:
case NUdf::TDataType<NUdf::TInterval64>::Id:
res.set_int64_value(value.GetInt64());
break;
case NUdf::TDataType<NUdf::TDecimal>::Id:
case NUdf::TDataType<NUdf::TUuid>::Id: {
res.set_low_128(value.GetLow128());
Expand Down Expand Up @@ -446,6 +454,34 @@ Y_FORCE_INLINE void ConvertData(NUdf::TDataTypeId typeId, const Ydb::Value& valu
}
res.SetInt64(value.int64_value());
break;
case NUdf::TDataType<NUdf::TDate32>::Id:
CheckTypeId(value.value_case(), Ydb::Value::kInt32Value, "Date");
if (value.int32_value() >= NUdf::MAX_DATE32) {
throw yexception() << "Invalid Date32 value";
}
res.SetInt32(value.int32_value());
break;
case NUdf::TDataType<NUdf::TDatetime64>::Id:
CheckTypeId(value.value_case(), Ydb::Value::kInt64Value, "Datetime");
if (value.int64_value() >= NUdf::MAX_DATETIME64) {
throw yexception() << "Invalid Datetime64 value";
}
res.SetInt64(value.int64_value());
break;
case NUdf::TDataType<NUdf::TTimestamp64>::Id:
CheckTypeId(value.value_case(), Ydb::Value::kInt64Value, "Timestamp");
if (value.int64_value() >= NUdf::MAX_TIMESTAMP64) {
throw yexception() << "Invalid Timestamp64 value";
}
res.SetInt64(value.int64_value());
break;
case NUdf::TDataType<NUdf::TInterval64>::Id:
CheckTypeId(value.value_case(), Ydb::Value::kInt64Value, "Interval");
if (std::abs(value.int64_value()) >= NUdf::MAX_INTERVAL64) {
throw yexception() << "Invalid Interval64 value";
}
res.SetInt64(value.int64_value());
break;
case NUdf::TDataType<NUdf::TUuid>::Id:
CheckTypeId(value.value_case(), Ydb::Value::kLow128, "Uuid");
res.SetLow128(value.low_128());
Expand Down Expand Up @@ -1075,6 +1111,22 @@ bool CheckValueData(NScheme::TTypeInfo type, const TCell& cell, TString& err) {
ok = (ui64)std::abs(cell.AsValue<i64>()) < NUdf::MAX_TIMESTAMP;
break;

case NScheme::NTypeIds::Date32:
ok = cell.AsValue<i32>() < NUdf::MAX_DATE32;
break;

case NScheme::NTypeIds::Datetime64:
ok = cell.AsValue<i64>() < NUdf::MAX_DATETIME64;
break;

case NScheme::NTypeIds::Timestamp64:
ok = cell.AsValue<i64>() < NUdf::MAX_TIMESTAMP64;
break;

case NScheme::NTypeIds::Interval64:
ok = std::abs(cell.AsValue<i64>()) < NUdf::MAX_INTERVAL64;
break;

case NScheme::NTypeIds::Utf8:
ok = NYql::IsUtf8(cell.AsBuf());
break;
Expand Down Expand Up @@ -1308,6 +1360,18 @@ void ProtoValueFromCell(NYdb::TValueBuilder& vb, const NScheme::TTypeInfo& typeI
case EPrimitiveType::Interval:
vb.Interval(cell.AsValue<i64>());
break;
case EPrimitiveType::Date32:
vb.Date32(cell.AsValue<i32>());
break;
case EPrimitiveType::Datetime64:
vb.Datetime64(cell.AsValue<i64>());
break;
case EPrimitiveType::Timestamp64:
vb.Timestamp64(cell.AsValue<i64>());
break;
case EPrimitiveType::Interval64:
vb.Interval64(cell.AsValue<i64>());
break;
case EPrimitiveType::TzDate:
vb.TzDate(getString());
break;
Expand Down

0 comments on commit ba35e13

Please sign in to comment.