check supported store types (#12568)

Author: zverevgeny

ydb/core/kqp/provider/yql_kikimr_type_ann.cpp

@@ -1256,9 +1256,17 @@ virtual TStatus HandleCreateTable(TKiCreateTable create, TExprContext& ctx) over
  "Can't reset TTL settings"));
  return TStatus::Error;
  } else if (name == "storeType") {
- TMaybe<TString> storeType = TString(setting.Value().Cast<TCoAtom>().Value());
- if (storeType && to_lower(storeType.GetRef()) == "column") {
- meta->StoreType = EStoreType::Column;
+ if (const TMaybe<TString> storeType = TString(setting.Value().Cast<TCoAtom>().Value())) {
+ const auto& val = to_lower(storeType.GetRef());
+ if (val == "column") {
+ meta->StoreType = EStoreType::Column;
+ } else if (val == "row") {
+ //pass
+ } else {
+ ctx.AddError(TIssue(ctx.GetPosition(setting.Name().Pos()),
+ TStringBuilder() << "Unsupported table store type: " << storeType.GetRef()));
+ return TStatus::Error;
+ }
  }
  } else if (name == "partitionByHashFunction") {
  meta->TableSettings.PartitionByHashFunction = TString(

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -4993,6 +4993,30 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
  }
  }
 
+ Y_UNIT_TEST(NEG_CreateTableWithUnsupportedStoreType) {
+ TKikimrSettings runnerSettings;
+ runnerSettings.WithSampleTables = false;
+ TKikimrRunner kikimr(runnerSettings);
+ auto db = kikimr.GetTableClient();
+ auto session = db.CreateSession().GetValueSync().GetSession();
+ TString tableStoreName = "/Root/TableStoreTest";
+ auto query = TStringBuilder() << R"(
+ --!syntax_v1
+ CREATE TABLE SomeTable (
+ Key Timestamp NOT NULL,
+ Value1 String,
+ Value2 Int64 NOT NULL,
+ PRIMARY KEY (Key)
+ )
+ WITH (
+ STORE = UNSUPPORTED
+ );
+ )"; 
+ auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+ UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::GENERIC_ERROR, result.GetIssues().ToString());
+ }
+
+
  Y_UNIT_TEST(CreateAlterDropTableStore) {
  TKikimrSettings runnerSettings;
  runnerSettings.WithSampleTables = false;