New CHANGEFEED's option: SCHEMA_CHANGES (#19303)

Author: CyberROFL

.github/last_commit.txt

@@ -1 +1 @@
-a19b867b82af456bce938a0506364f0ba85214a5
+8bbfc1965ae5de65c96c36238b45ec33dd84f0a0

include/ydb-cpp-sdk/client/table/table.h

@@ -397,6 +397,8 @@ class TChangefeedDescription {
 
     // Enable virtual timestamps
     TChangefeedDescription& WithVirtualTimestamps();
+    // Enable schema changes
+    TChangefeedDescription& WithSchemaChanges();
     // Enable resolved timestamps
     TChangefeedDescription& WithResolvedTimestamps(const TDuration& interval);
     // Customise retention period of underlying topic (24h by default).
@@ -415,6 +417,7 @@ class TChangefeedDescription {
     EChangefeedFormat GetFormat() const;
     EChangefeedState GetState() const;
     bool GetVirtualTimestamps() const;
+    bool GetSchemaChanges() const;
     const std::optional<TDuration>& GetResolvedTimestamps() const;
     bool GetInitialScan() const;
     const std::unordered_map<std::string, std::string>& GetAttributes() const;
@@ -442,6 +445,7 @@ class TChangefeedDescription {
     EChangefeedFormat Format_;
     EChangefeedState State_ = EChangefeedState::Unknown;
     bool VirtualTimestamps_ = false;
+    bool SchemaChanges_ = false;
     std::optional<TDuration> ResolvedTimestamps_;
     std::optional<TDuration> RetentionPeriod_;
     bool InitialScan_ = false;

src/api/protos/ydb_table.proto

@@ -233,6 +233,8 @@ message Changefeed {
     google.protobuf.Duration resolved_timestamps_interval = 9;
     // Partitioning settings of underlying topic.
     Topic.PartitioningSettings topic_partitioning_settings = 10;
+    // Emit schema change events or not
+    bool schema_changes = 11;
 }
 
 message ChangefeedDescription {
@@ -274,6 +276,8 @@ message ChangefeedDescription {
     google.protobuf.Duration resolved_timestamps_interval = 8;
     // Progress of initial scan. If unspecified, initial scan was not launched.
     InitialScanProgress initial_scan_progress = 9;
+    // State of emitting of schema change events
+    bool schema_changes = 10;
 }
 
 message StoragePool {

src/client/table/table.cpp

@@ -2687,6 +2687,11 @@ TChangefeedDescription& TChangefeedDescription::WithVirtualTimestamps() {
     return *this;
 }
 
+TChangefeedDescription& TChangefeedDescription::WithSchemaChanges() {
+    SchemaChanges_ = true;
+    return *this;
+}
+
 TChangefeedDescription& TChangefeedDescription::WithResolvedTimestamps(const TDuration& value) {
     ResolvedTimestamps_ = value;
     return *this;
@@ -2742,6 +2747,10 @@ bool TChangefeedDescription::GetVirtualTimestamps() const {
     return VirtualTimestamps_;
 }
 
+bool TChangefeedDescription::GetSchemaChanges() const {
+    return SchemaChanges_;
+}
+
 const std::optional<TDuration>& TChangefeedDescription::GetResolvedTimestamps() const {
     return ResolvedTimestamps_;
 }
@@ -2806,6 +2815,9 @@ TChangefeedDescription TChangefeedDescription::FromProto(const TProto& proto) {
     if (proto.virtual_timestamps()) {
         ret.WithVirtualTimestamps();
     }
+    if (proto.schema_changes()) {
+        ret.WithSchemaChanges();
+    }
     if (proto.has_resolved_timestamps_interval()) {
         ret.WithResolvedTimestamps(TDuration::MilliSeconds(
             ::google::protobuf::util::TimeUtil::DurationToMilliseconds(proto.resolved_timestamps_interval())));
@@ -2849,6 +2861,7 @@ template <typename TProto>
 void TChangefeedDescription::SerializeCommonFields(TProto& proto) const {
     proto.set_name(TStringType{Name_});
     proto.set_virtual_timestamps(VirtualTimestamps_);
+    proto.set_schema_changes(SchemaChanges_);
     proto.set_aws_region(TStringType{AwsRegion_});
 
     switch (Mode_) {
@@ -2932,7 +2945,8 @@ void TChangefeedDescription::Out(IOutputStream& o) const {
     o << "{ name: \"" << Name_ << "\""
       << ", mode: " << Mode_ << ""
       << ", format: " << Format_ << ""
-      << ", virtual_timestamps: " << (VirtualTimestamps_ ? "on": "off") << "";
+      << ", virtual_timestamps: " << (VirtualTimestamps_ ? "on": "off") << ""
+      << ", schema_changes: " << (SchemaChanges_ ? "on": "off") << "";
 
     if (ResolvedTimestamps_) {
         o << ", resolved_timestamps: " << *ResolvedTimestamps_;
@@ -2958,6 +2972,7 @@ bool operator==(const TChangefeedDescription& lhs, const TChangefeedDescription&
         && lhs.GetMode() == rhs.GetMode()
         && lhs.GetFormat() == rhs.GetFormat()
         && lhs.GetVirtualTimestamps() == rhs.GetVirtualTimestamps()
+        && lhs.GetSchemaChanges() == rhs.GetSchemaChanges()
         && lhs.GetResolvedTimestamps() == rhs.GetResolvedTimestamps()
         && lhs.GetAwsRegion() == rhs.GetAwsRegion();
 }