Invalidate query compilation cache entries with outdated VIEWs (#1960)

ydb/core/kqp/provider/yql_kikimr_datasource.cpp

@@ -351,6 +351,18 @@ class TKiSourceLoadTableMetadataTransformer : public TGraphTransformerBase {
                 if (!AddCluster(table, res, input, ctx)) {
                     return TStatus::Error;
                 }
+
+                if (const auto& preparingQuery = SessionCtx->Query().PreparingQuery;
+                        preparingQuery
+                        && res.Metadata->Kind == EKikimrTableKind::View
+                ) {
+                    const auto& viewMetadata = *res.Metadata;
+                    auto* viewInfo = preparingQuery->MutablePhysicalQuery()->MutableViewInfos()->Add();
+                    auto* pathId = viewInfo->MutableTableId();
+                    pathId->SetOwnerId(viewMetadata.PathId.OwnerId());
+                    pathId->SetTableId(viewMetadata.PathId.TableId());
+                    viewInfo->SetSchemaVersion(viewMetadata.SchemaVersion);
+                }
             } else {
                 TIssueScopeGuard issueScope(ctx.IssueManager, [input, &table, &ctx]() {
                     return MakeIntrusive<TIssue>(TIssue(ctx.GetPosition(input->Pos()), TStringBuilder()

ydb/core/kqp/session_actor/kqp_query_state.cpp

@@ -71,13 +71,24 @@ bool TKqpQueryState::EnsureTableVersions(const TEvTxProxySchemeCache::TEvNavigat
     return true;
 }
 
+void TKqpQueryState::FillViews(const google::protobuf::RepeatedPtrField< ::NKqpProto::TKqpTableInfo>& views) {
+    for (const auto& view : views) {
+        const auto& pathId = view.GetTableId();
+        const auto schemaVersion = view.GetSchemaVersion();
+        auto [it, isInserted] = TableVersions.emplace(TTableId(pathId.GetOwnerId(), pathId.GetTableId()), schemaVersion);
+        if (!isInserted) {
+            Y_ENSURE(it->second == schemaVersion);
+        }
+    }
+}
 
 std::unique_ptr<TEvTxProxySchemeCache::TEvNavigateKeySet> TKqpQueryState::BuildNavigateKeySet() {
     TableVersions.clear();
 
     for (const auto& tx : PreparedQuery->GetPhysicalQuery().GetTransactions()) {
         FillTables(tx);
     }
+    FillViews(PreparedQuery->GetPhysicalQuery().GetViewInfos());
 
     auto navigate = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
     navigate->DatabaseName = Database;

ydb/core/kqp/session_actor/kqp_query_state.h

@@ -234,6 +234,8 @@ class TKqpQueryState : public TNonCopyable {
         }
     }
 
+    void FillViews(const google::protobuf::RepeatedPtrField< ::NKqpProto::TKqpTableInfo>& views);
+
     bool NeedCheckTableVersions() const {
         return CompileStats.FromCache;
     }

ydb/core/kqp/ut/view/view_ut.cpp

@@ -1,13 +1,15 @@
 #include <ydb/core/kqp/ut/common/kqp_ut_common.h>
 #include <ydb/library/yql/sql/sql.h>
 #include <ydb/library/yql/utils/log/log.h>
+#include <ydb/public/sdk/cpp/client/ydb_proto/accessor.h>
 
 #include <util/folder/filelist.h>
 
 #include <format>
 
 using namespace NKikimr;
 using namespace NKikimr::NKqp;
+using namespace NYdb;
 using namespace NYdb::NTable;
 
 namespace {
@@ -60,17 +62,46 @@ void ExecuteDataDefinitionQuery(TSession& session, const TString& script) {
                                           << script << "\nThe issues:\n" << result.GetIssues().ToString());
 }
 
-TDataQueryResult ExecuteDataModificationQuery(TSession& session, const TString& script) {
+TDataQueryResult ExecuteDataModificationQuery(TSession& session,
+                                              const TString& script,
+                                              const TExecDataQuerySettings& settings = {}
+) {
     const auto result = session.ExecuteDataQuery(
             script,
-            TTxControl::BeginTx(TTxSettings::SerializableRW()).CommitTx()
+            TTxControl::BeginTx(TTxSettings::SerializableRW()).CommitTx(),
+            settings
         ).ExtractValueSync();
     UNIT_ASSERT_C(result.IsSuccess(), "Failed to execute the following DML script:\n"
                                           << script << "\nThe issues:\n" << result.GetIssues().ToString());
 
     return result;
 }
 
+TValue GetSingleResult(const TDataQueryResult& rawResults) {
+    auto resultSetParser = rawResults.GetResultSetParser(0);
+    UNIT_ASSERT(resultSetParser.TryNextRow());
+    return resultSetParser.GetValue(0);
+}
+
+int GetInteger(const TValue& value) {
+    return TValueParser(value).GetInt32();
+}
+
+TMaybe<bool> GetFromCacheStat(const TQueryStats& stats) {
+    const auto& proto = TProtoAccessor::GetProto(stats);
+    if (!proto.Hascompilation()) {
+        return Nothing();
+    }
+    return proto.Getcompilation().Getfrom_cache();
+}
+
+void AssertFromCache(const TMaybe<TQueryStats>& stats, bool expectedValue) {
+    UNIT_ASSERT(stats.Defined());
+    const auto isFromCache = GetFromCacheStat(*stats);
+    UNIT_ASSERT_C(isFromCache.Defined(), stats->ToString());
+    UNIT_ASSERT_VALUES_EQUAL_C(*isFromCache, expectedValue, stats->ToString());
+}
+
 void CompareResults(const TDataQueryResult& first, const TDataQueryResult& second) {
     const auto& firstResults = first.GetResultSets();
     const auto& secondResults = second.GetResultSets();
@@ -376,4 +407,51 @@ Y_UNIT_TEST_SUITE(TSelectFromViewTest) {
             ExecuteDataDefinitionQuery(session, ReadWholeFile(pathPrefix + "drop_view.sql"));
         }
     }
+
+    Y_UNIT_TEST(QueryCacheIsUpdated) {
+        TKikimrRunner kikimr(TKikimrSettings().SetWithSampleTables(false));
+        EnableViewsFeatureFlag(kikimr);
+        auto session = kikimr.GetTableClient().CreateSession().GetValueSync().GetSession();
+
+        constexpr const char* viewName = "TheView";
+
+        const auto getCreationQuery = [&viewName](const char* innerQuery) -> TString {
+            return std::format(R"(
+                    CREATE VIEW {} WITH (security_invoker = TRUE) AS {};
+                )",
+                viewName,
+                innerQuery
+            );
+        };
+        constexpr const char* firstInnerQuery = "SELECT 1";
+        ExecuteDataDefinitionQuery(session, getCreationQuery(firstInnerQuery));
+
+        const TString selectFromViewQuery = std::format(R"(
+                SELECT * FROM {};
+            )",
+            viewName
+        );
+        TExecDataQuerySettings queryExecutionSettings;
+        queryExecutionSettings.KeepInQueryCache(true);
+        queryExecutionSettings.CollectQueryStats(ECollectQueryStatsMode::Basic);
+        ExecuteDataModificationQuery(session, selectFromViewQuery, queryExecutionSettings);
+        // make sure the server side cache is working by calling the same query twice
+        const auto cachedQueryRawResult = ExecuteDataModificationQuery(session, selectFromViewQuery, queryExecutionSettings);
+        AssertFromCache(cachedQueryRawResult.GetStats(), true);
+        UNIT_ASSERT_VALUES_EQUAL(GetInteger(GetSingleResult(cachedQueryRawResult)), 1);
+
+        // recreate the view with a different query inside
+        ExecuteDataDefinitionQuery(session, std::format(R"(
+                    DROP VIEW {};
+                )",
+                viewName
+            )
+        );
+        constexpr const char* secondInnerQuery = "SELECT 2";
+        ExecuteDataDefinitionQuery(session, getCreationQuery(secondInnerQuery));
+
+        const auto secondCallRawResult = ExecuteDataModificationQuery(session, selectFromViewQuery, queryExecutionSettings);
+        AssertFromCache(secondCallRawResult.GetStats(), false);
+        UNIT_ASSERT_VALUES_EQUAL(GetInteger(GetSingleResult(secondCallRawResult)), 2);
+    }
 }

ydb/core/protos/kqp_physical.proto

@@ -507,4 +507,6 @@ message TKqpPhyQuery {
     bool HasUncommittedChangesRead = 9;
 
     string QueryDiagnostics = 10;
+
+    repeated TKqpTableInfo ViewInfos = 11;
 }