Diff schemas (ydb-platform#10958)

zverevgeny · Jan 1, 2025 · abefcc9 · abefcc9
1 parent b1f6580
commit abefcc9
Show file tree

Hide file tree

Showing 37 changed files with 604 additions and 305 deletions.
diff --git a/ydb/core/formats/arrow/splitter/scheme_info.h b/ydb/core/formats/arrow/splitter/scheme_info.h
@@ -22,8 +22,6 @@ class ISchemaDetailInfo {
     NAccessor::TColumnSaver GetColumnSaver(const ui32 columnId) const;
     virtual std::shared_ptr<arrow::Field> GetField(const ui32 columnId) const = 0;
     virtual std::optional<TColumnSerializationStat> GetColumnSerializationStats(const ui32 columnId) const = 0;
-    virtual bool NeedMinMaxForColumn(const ui32 columnId) const = 0;
-    virtual bool IsSortedColumn(const ui32 columnId) const = 0;
     virtual std::optional<TBatchSerializationStat> GetBatchSerializationStats(const std::shared_ptr<arrow::RecordBatch>& rb) const = 0;
 };
 }   // namespace NKikimr::NArrow::NSplitter
diff --git a/ydb/core/grpc_services/rpc_create_table.cpp b/ydb/core/grpc_services/rpc_create_table.cpp
@@ -109,7 +109,6 @@ class TCreateTableRPC : public TRpcSchemeRequestActor<TCreateTableRPC, TEvCreate
         tableDesc->SetName(tableName);
 
         auto schema = tableDesc->MutableSchema();
-        schema->SetEngine(NKikimrSchemeOp::EColumnTableEngine::COLUMN_ENGINE_REPLACING_TIMESERIES);
 
         TString error;
         if (!FillColumnDescription(*tableDesc, req.columns(), code, error)) {

diff --git a/ydb/core/grpc_services/rpc_log_store.cpp b/ydb/core/grpc_services/rpc_log_store.cpp
@@ -120,19 +120,13 @@ bool ConvertSchemaFromPublicToInternal(const Ydb::LogStore::Schema& from, NKikim
         }
     }
 
-    to.SetEngine(NKikimrSchemeOp::COLUMN_ENGINE_REPLACING_TIMESERIES);
     status = {};
     return true;
 }
 
 bool ConvertSchemaFromInternalToPublic(const NKikimrSchemeOp::TColumnTableSchema& from, Ydb::LogStore::Schema& to,
     Ydb::StatusIds::StatusCode& status, TString& error)
 {
-    if (from.GetEngine() != NKikimrSchemeOp::COLUMN_ENGINE_REPLACING_TIMESERIES) {
-        status = Ydb::StatusIds::INTERNAL_ERROR;
-        error = TStringBuilder() << "Unexpected table engine: " << NKikimrSchemeOp::EColumnTableEngine_Name(from.GetEngine());
-        return false;
-    }
     to.mutable_primary_key()->CopyFrom(from.GetKeyColumnNames());
     for (const auto& column : from.GetColumns()) {
         auto* col = to.add_columns();

diff --git a/ydb/core/kqp/host/kqp_gateway_proxy.cpp b/ydb/core/kqp/host/kqp_gateway_proxy.cpp
@@ -393,8 +393,6 @@ void FillColumnTableSchema(NKikimrSchemeOp::TColumnTableSchema& schema, const T&
     for (const auto& keyColumn : metadata.KeyColumnNames) {
         schema.AddKeyColumnNames(keyColumn);
     }
-
-    schema.SetEngine(NKikimrSchemeOp::EColumnTableEngine::COLUMN_ENGINE_REPLACING_TIMESERIES);
 }
 
 bool FillCreateColumnTableDesc(NYql::TKikimrTableMetadataPtr metadata,

diff --git a/ydb/core/kqp/ut/olap/helpers/typed_local.cpp b/ydb/core/kqp/ut/olap/helpers/typed_local.cpp
@@ -13,7 +13,6 @@ TString TTypedLocalHelper::GetTestTableSchema() const {
             Columns { Name: "pk_int" Type: "Int64" NotNull: true }
             Columns { Name: "ts" Type: "Timestamp" }
             KeyColumnNames: "pk_int"
-            Engine: COLUMN_ENGINE_REPLACING_TIMESERIES
         )";
     return result;
 }
@@ -34,7 +33,6 @@ TString TTypedLocalHelper::GetMultiColumnTestTableSchema(ui32 reps) const {
     }
     result += R"(
             KeyColumnNames: "pk_int"
-            Engine: COLUMN_ENGINE_REPLACING_TIMESERIES
     )";
     return result;
 }

diff --git a/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp b/ydb/core/kqp/ut/olap/kqp_olap_ut.cpp
@@ -116,7 +116,6 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
                                              Columns { Name: "Datetime_column" Type: "Datetime" }
                                              #Columns { Name: "Interval_column" Type: "Interval" }
                                              KeyColumnNames: "key"
-                                             Engine: COLUMN_ENGINE_REPLACING_TIMESERIES
                                          }
                                      }
         )");

diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -556,9 +556,6 @@ message TColumnTableSchema {
     repeated TOlapColumnDescription Columns = 1;
     repeated string KeyColumnNames = 2;
 
-    // A type of engine used by the table
-    optional EColumnTableEngine Engine = 3;
-
     // Internal fields
     optional uint32 NextColumnId = 4;
 
@@ -569,11 +566,24 @@ message TColumnTableSchema {
     //optional int32 DefaultCompressionLevel = 7; // deprecated, not used before replace
     optional TCompressionOptions DefaultCompression = 8;
 
-    optional bool CompositeMarksDeprecated = 9 [ default = false ];
     repeated TOlapIndexDescription Indexes = 10;
     optional TColumnTableSchemeOptions Options = 12;
 }
 
+message TColumnTableSchemaDiff {
+    optional uint64 Version = 1;
+
+    repeated TOlapColumnDescription UpsertColumns = 2;
+    repeated uint32 DropColumns = 3;
+
+    optional TCompressionOptions DefaultCompression = 4;
+
+    repeated TOlapIndexDescription UpsertIndexes = 5;
+    repeated uint32 DropIndexes = 6;
+
+    optional TColumnTableSchemeOptions Options = 7;
+}
+
 message TColumnTableRequestedOptions {
     optional bool SchemeNeedActualization = 1 [default = false];
     optional bool ExternalGuaranteeExclusivePK = 2;

diff --git a/ydb/core/protos/tx_columnshard.proto b/ydb/core/protos/tx_columnshard.proto
@@ -237,6 +237,7 @@ message TSchemaPresetVersionInfo {
     optional uint64 SinceStep = 2;
     optional uint64 SinceTxId = 3;
     optional NKikimrSchemeOp.TColumnTableSchema Schema = 4;
+    optional NKikimrSchemeOp.TColumnTableSchemaDiff Diff = 5;
 }
 
 message TTtlSettingsPresetVersionInfo {

diff --git a/ydb/core/testlib/cs_helper.cpp b/ydb/core/testlib/cs_helper.cpp
@@ -193,7 +193,6 @@ TString THelper::GetTestTableSchema() const {
     sb << R"(
         KeyColumnNames: "timestamp"
         KeyColumnNames: "uid"
-        Engine : COLUMN_ENGINE_REPLACING_TIMESERIES
     )";
     return sb;
 }

diff --git a/ydb/core/testlib/cs_helper.h b/ydb/core/testlib/cs_helper.h
@@ -53,7 +53,6 @@ class THelper: public THelperSchemaless {
         Columns { Name: "level" Type: "Int32" DataAccessorConstructor{ ClassName: "SPARSED" }}
         Columns { Name: "message" Type: "Utf8" }
         KeyColumnNames: "timestamp"
-        Engine: COLUMN_ENGINE_REPLACING_TIMESERIES
     )";
 
     void WithSomeNulls() {

diff --git a/ydb/core/tx/columnshard/engines/column_engine.h b/ydb/core/tx/columnshard/engines/column_engine.h
@@ -264,6 +264,34 @@ class IColumnEngine {
     virtual void DoRegisterTable(const ui64 pathId) = 0;
 
 public:
+    class TSchemaInitializationData {
+    private:
+        YDB_READONLY_DEF(std::optional<NKikimrSchemeOp::TColumnTableSchema>, Schema);
+        YDB_READONLY_DEF(std::optional<NKikimrSchemeOp::TColumnTableSchemaDiff>, Diff);
+
+    public:
+        const NKikimrSchemeOp::TColumnTableSchema& GetSchemaVerified() const {
+            AFL_VERIFY(Schema);
+            return *Schema;
+        }
+
+        TSchemaInitializationData(
+            const std::optional<NKikimrSchemeOp::TColumnTableSchema>& schema, const std::optional<NKikimrSchemeOp::TColumnTableSchemaDiff>& diff)
+            : Schema(schema)
+            , Diff(diff) {
+            AFL_VERIFY(Schema || Diff);
+        }
+
+        TSchemaInitializationData(const NKikimrTxColumnShard::TSchemaPresetVersionInfo& info) {
+            if (info.HasSchema()) {
+                Schema = info.GetSchema();
+            }
+            if (info.HasDiff()) {
+                Diff = info.GetDiff();
+            }
+        }
+    };
+
     static ui64 GetMetadataLimit();
 
     virtual ~IColumnEngine() = default;
@@ -293,7 +321,7 @@ class IColumnEngine {
     virtual bool ApplyChangesOnTxCreate(std::shared_ptr<TColumnEngineChanges> changes, const TSnapshot& snapshot) noexcept = 0;
     virtual bool ApplyChangesOnExecute(IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> changes, const TSnapshot& snapshot) noexcept = 0;
     virtual void RegisterSchemaVersion(const TSnapshot& snapshot, TIndexInfo&& info) = 0;
-    virtual void RegisterSchemaVersion(const TSnapshot& snapshot, const NKikimrSchemeOp::TColumnTableSchema& schema) = 0;
+    virtual void RegisterSchemaVersion(const TSnapshot& snapshot, const TSchemaInitializationData& schema) = 0;
     virtual const TMap<ui64, std::shared_ptr<TColumnEngineStats>>& GetStats() const = 0;
     virtual const TColumnEngineStats& GetTotalStats() = 0;
     virtual ui64 MemoryUsage() const {

diff --git a/ydb/core/tx/columnshard/engines/column_engine_logs.cpp b/ydb/core/tx/columnshard/engines/column_engine_logs.cpp
@@ -26,8 +26,8 @@
 
 namespace NKikimr::NOlap {
 
-TColumnEngineForLogs::TColumnEngineForLogs(ui64 tabletId, const std::shared_ptr<IStoragesManager>& storagesManager,
-    const TSnapshot& snapshot, const NKikimrSchemeOp::TColumnTableSchema& schema)
+TColumnEngineForLogs::TColumnEngineForLogs(
+    ui64 tabletId, const std::shared_ptr<IStoragesManager>& storagesManager, const TSnapshot& snapshot, const TSchemaInitializationData& schema)
     : GranulesStorage(std::make_shared<TGranulesStorage>(SignalCounters, storagesManager))
     , StoragesManager(storagesManager)
     , TabletId(tabletId)
@@ -162,11 +162,17 @@ void TColumnEngineForLogs::RegisterSchemaVersion(const TSnapshot& snapshot, TInd
     }
 }
 
-void TColumnEngineForLogs::RegisterSchemaVersion(const TSnapshot& snapshot, const NKikimrSchemeOp::TColumnTableSchema& schema) {
-    std::optional<NOlap::TIndexInfo> indexInfoOptional = NOlap::TIndexInfo::BuildFromProto(schema, StoragesManager, SchemaObjectsCache);
+void TColumnEngineForLogs::RegisterSchemaVersion(const TSnapshot& snapshot, const TSchemaInitializationData& schema) {
+    std::optional<NOlap::TIndexInfo> indexInfoOptional;
+    if (schema.GetDiff()) {
+        AFL_VERIFY(!VersionedIndex.IsEmpty());
+        indexInfoOptional = NOlap::TIndexInfo::BuildFromProto(
+            *schema.GetDiff(), VersionedIndex.GetLastSchema()->GetIndexInfo(), StoragesManager, SchemaObjectsCache);
+    } else {
+        indexInfoOptional = NOlap::TIndexInfo::BuildFromProto(schema.GetSchemaVerified(), StoragesManager, SchemaObjectsCache);
+    }
     AFL_VERIFY(indexInfoOptional);
-    NOlap::TIndexInfo indexInfo = std::move(*indexInfoOptional);
-    RegisterSchemaVersion(snapshot, std::move(indexInfo));
+    RegisterSchemaVersion(snapshot, std::move(*indexInfoOptional));
 }
 
 bool TColumnEngineForLogs::Load(IDbWrapper& db) {

diff --git a/ydb/core/tx/columnshard/engines/column_engine_logs.h b/ydb/core/tx/columnshard/engines/column_engine_logs.h
@@ -83,7 +83,7 @@ class TColumnEngineForLogs: public IColumnEngine {
     };
 
     TColumnEngineForLogs(ui64 tabletId, const std::shared_ptr<IStoragesManager>& storagesManager, const TSnapshot& snapshot,
-        const NKikimrSchemeOp::TColumnTableSchema& schema);
+        const TSchemaInitializationData& schema);
     TColumnEngineForLogs(
         ui64 tabletId, const std::shared_ptr<IStoragesManager>& storagesManager, const TSnapshot& snapshot, TIndexInfo&& schema);
 
@@ -131,7 +131,7 @@ class TColumnEngineForLogs: public IColumnEngine {
         IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> indexChanges, const TSnapshot& snapshot) noexcept override;
 
     void RegisterSchemaVersion(const TSnapshot& snapshot, TIndexInfo&& info) override;
-    void RegisterSchemaVersion(const TSnapshot& snapshot, const NKikimrSchemeOp::TColumnTableSchema& schema) override;
+    void RegisterSchemaVersion(const TSnapshot& snapshot, const TSchemaInitializationData& schema) override;
 
     std::shared_ptr<TSelectInfo> Select(
         ui64 pathId, TSnapshot snapshot, const TPKRangesFilter& pkRangesFilter, const bool withUncommitted) const override;

diff --git a/ydb/core/tx/columnshard/engines/scheme/abstract/index_info.h b/ydb/core/tx/columnshard/engines/scheme/abstract/index_info.h
@@ -31,6 +31,7 @@ class IIndexInfo {
     static constexpr const char* SPEC_COL_TX_ID = NOlap::NPortion::TSpecialColumns::SPEC_COL_TX_ID;
     static constexpr const char* SPEC_COL_WRITE_ID = NOlap::NPortion::TSpecialColumns::SPEC_COL_WRITE_ID;
     static constexpr const char* SPEC_COL_DELETE_FLAG = NOlap::NPortion::TSpecialColumns::SPEC_COL_DELETE_FLAG;
+    static constexpr ui32 SpecialColumnsCount = 4;
 
     static const inline std::shared_ptr<arrow::Field> PlanStepField = arrow::field(SPEC_COL_PLAN_STEP, arrow::uint64());
     static const inline std::shared_ptr<arrow::Field> TxIdField = arrow::field(SPEC_COL_TX_ID, arrow::uint64());