YT-20983: Add dynamic tables versioned map reduce read

dave11ar · dave11ar · commit 21704936a7af · 2024-08-22T12:19:27.000+03:00
12c8b4a9cd18c32a89bb2bc6d78fd4dc0f1da7b5
diff --git a/yt/yt/client/api/rpc_proxy/helpers.cpp b/yt/yt/client/api/rpc_proxy/helpers.cpp
@@ -619,22 +619,6 @@ void FromProto(
     FromProto(&statistics->InnerStatistics, protoStatistics.inner_statistics());
 }
 
-void ToProto(
-    NProto::TVersionedReadOptions* protoOptions,
-    const NTableClient::TVersionedReadOptions& options)
-{
-    protoOptions->set_read_mode(static_cast<i32>(options.ReadMode));
-}
-
-void FromProto(
-    NTableClient::TVersionedReadOptions* options,
-    const NProto::TVersionedReadOptions& protoOptions)
-{
-    if (protoOptions.has_read_mode()) {
-        options->ReadMode = CheckedEnumCast<EVersionedIOMode>(protoOptions.read_mode());
-    }
-}
-
 void ToProto(NProto::TOperation* protoOperation, const NApi::TOperation& operation)
 {
     protoOperation->Clear();
diff --git a/yt/yt/client/api/rpc_proxy/helpers.h b/yt/yt/client/api/rpc_proxy/helpers.h
@@ -136,14 +136,6 @@ void FromProto(
     NQueryClient::TQueryStatistics* statistics,
     const NProto::TQueryStatistics& protoStatistics);
 
-void ToProto(
-    NProto::TVersionedReadOptions* protoOptions,
-    const NTableClient::TVersionedReadOptions& options);
-
-void FromProto(
-    NTableClient::TVersionedReadOptions* options,
-    const NProto::TVersionedReadOptions& protoOptions);
-
 void ToProto(
     NProto::TOperation* protoOperation,
     const NApi::TOperation& operation);
diff --git a/yt/yt/client/table_client/public.h b/yt/yt/client/table_client/public.h
@@ -46,6 +46,7 @@ class THunkChunkRef;
 class TColumnMetaExt;
 class TVersionedRowDigestExt;
 class TCompressionDictionaryExt;
+class TVersionedReadOptions;
 
 } // namespace NProto
 
diff --git a/yt/yt/client/table_client/schema.cpp b/yt/yt/client/table_client/schema.cpp
@@ -799,7 +799,7 @@ TKeyColumns TTableSchema::GetKeyColumns() const
 
 int TTableSchema::GetColumnCount() const
 {
-    return static_cast<int>(Columns().size());
+    return ssize(Columns());
 }
 
 std::vector<TString> TTableSchema::GetColumnNames() const
@@ -1739,7 +1739,8 @@ void ValidateDynamicTableKeyColumnCount(int count)
 void ValidateSystemColumnSchema(
     const TColumnSchema& columnSchema,
     bool isTableSorted,
-    bool allowUnversionedUpdateColumns)
+    bool allowUnversionedUpdateColumns,
+    bool allowTimestampColumns)
 {
     static const auto allowedSortedTablesSystemColumns = THashMap<TString, ESimpleLogicalValueType>{
         {EmptyValueColumnName, ESimpleLogicalValueType::Int64},
@@ -1793,6 +1794,13 @@ void ValidateSystemColumnSchema(
         }
     }
 
+    if (allowTimestampColumns) {
+        if (name.StartsWith(TimestampColumnPrefix)) {
+            validateType(ESimpleLogicalValueType::Uint64);
+            return;
+        }
+    }
+
     // Unexpected system column.
     THROW_ERROR_EXCEPTION("System column name %Qv is not allowed here",
         name);
@@ -1816,7 +1824,8 @@ void ValidateColumnSchema(
     const TColumnSchema& columnSchema,
     bool isTableSorted,
     bool isTableDynamic,
-    bool allowUnversionedUpdateColumns)
+    bool allowUnversionedUpdateColumns,
+    bool allowTimestampColumns)
 {
     static const auto allowedAggregates = THashSet<TString>{
         "sum",
@@ -1842,7 +1851,11 @@ void ValidateColumnSchema(
         ValidateColumnName(name);
 
         if (stableName.Underlying().StartsWith(SystemColumnNamePrefix) || name.StartsWith(SystemColumnNamePrefix)) {
-            ValidateSystemColumnSchema(columnSchema, isTableSorted, allowUnversionedUpdateColumns);
+            ValidateSystemColumnSchema(
+                columnSchema,
+                isTableSorted,
+                allowUnversionedUpdateColumns,
+                allowTimestampColumns);
         }
 
         {
@@ -2173,15 +2186,20 @@ void ValidateSchemaAttributes(const TTableSchema& schema)
     }
 }
 
-void ValidateTableSchema(const TTableSchema& schema, bool isTableDynamic, bool allowUnversionedUpdateColumns)
+void ValidateTableSchema(
+    const TTableSchema& schema,
+    bool isTableDynamic,
+    bool allowUnversionedUpdateColumns,
+    bool allowTimestampColumns)
 {
     int totalTypeComplexity = 0;
     for (const auto& column : schema.Columns()) {
         ValidateColumnSchema(
             column,
             schema.IsSorted(),
             isTableDynamic,
-            allowUnversionedUpdateColumns);
+            allowUnversionedUpdateColumns,
+            allowTimestampColumns);
         if (!schema.GetStrict() && column.IsRenamed()) {
             THROW_ERROR_EXCEPTION("Renamed column %v in non-strict schema",
                 column.GetDiagnosticNameString());
diff --git a/yt/yt/client/table_client/schema.h b/yt/yt/client/table_client/schema.h
@@ -500,12 +500,14 @@ void ValidateColumnSchema(
     const TColumnSchema& columnSchema,
     bool isTableSorted = false,
     bool isTableDynamic = false,
-    bool allowUnversionedUpdateColumns = false);
+    bool allowUnversionedUpdateColumns = false,
+    bool allowTimestampColumns = false);
 
 void ValidateTableSchema(
     const TTableSchema& schema,
     bool isTableDynamic = false,
-    bool allowUnversionedUpdateColumns = false);
+    bool allowUnversionedUpdateColumns = false,
+    bool allowTimestampColumns = false);
 
 void ValidateNoDescendingSortOrder(const TTableSchema& schema);
 
diff --git a/yt/yt/client/table_client/versioned_io_options.cpp b/yt/yt/client/table_client/versioned_io_options.cpp
@@ -1,5 +1,7 @@
 #include "versioned_io_options.h"
 
+#include <yt_proto/yt/client/table_client/proto/versioned_io_options.pb.h>
+
 namespace NYT::NTableClient {
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -10,6 +12,20 @@ void TVersionedReadOptions::Register(TRegistrar registrar)
         .Default(EVersionedIOMode::Default);
 }
 
+void ToProto(
+    NProto::TVersionedReadOptions* protoOptions,
+    const TVersionedReadOptions& options)
+{
+    protoOptions->set_read_mode(static_cast<i32>(options.ReadMode));
+}
+
+void FromProto(
+    TVersionedReadOptions* options,
+    const NProto::TVersionedReadOptions& protoOptions)
+{
+    options->ReadMode = CheckedEnumCast<EVersionedIOMode>(protoOptions.read_mode());
+}
+
 std::optional<TString> GetTimestampColumnOriginalNameOrNull(TStringBuf name)
 {
     auto prefixEnd = name.begin() + ssize(TimestampColumnPrefix);
diff --git a/yt/yt/client/table_client/versioned_io_options.h b/yt/yt/client/table_client/versioned_io_options.h
@@ -23,6 +23,14 @@ struct TVersionedReadOptions
     static void Register(TRegistrar registrar);
 };
 
+void ToProto(
+    NProto::TVersionedReadOptions* protoOptions,
+    const NTableClient::TVersionedReadOptions& options);
+
+void FromProto(
+    NTableClient::TVersionedReadOptions* options,
+    const NProto::TVersionedReadOptions& protoOptions);
+
 std::optional<TString> GetTimestampColumnOriginalNameOrNull(TStringBuf name);
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/client/ypath/rich.cpp b/yt/yt/client/ypath/rich.cpp
@@ -7,6 +7,7 @@
 #include <yt/yt/client/table_client/column_sort_schema.h>
 #include <yt/yt/client/table_client/column_rename_descriptor.h>
 #include <yt/yt/client/table_client/schema.h>
+#include <yt/yt/client/table_client/versioned_io_options.h>
 
 #include <yt/yt/core/misc/error.h>
 
@@ -679,6 +680,11 @@ bool TRichYPath::GetCreate() const
     return GetAttribute<bool>(*this, "create", false);
 }
 
+TVersionedReadOptions TRichYPath::GetVersionedReadOptions() const
+{
+    return GetAttribute(*this, "versioned_read_options", TVersionedReadOptions());
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 TString ConvertToString(const TRichYPath& path, EYsonFormat ysonFormat)
@@ -779,6 +785,7 @@ const std::vector<TString>& GetWellKnownRichYPathAttributes()
         "clusters",
         "create",
         "read_via_exec_node",
+        "versioned_read_options",
     };
     return WellKnownAttributes;
 }
diff --git a/yt/yt/client/ypath/rich.h b/yt/yt/client/ypath/rich.h
@@ -2,22 +2,22 @@
 
 #include "public.h"
 
-#include <yt/yt/core/yson/public.h>
-
-#include <yt/yt/core/ytree/attributes.h>
-
-#include <yt/yt/core/compression/public.h>
+#include <yt/yt/client/chunk_client/read_limit.h>
 
-#include <yt/yt/library/erasure/public.h>
+#include <yt/yt/client/security_client/public.h>
 
 #include <yt/yt/client/table_client/column_sort_schema.h>
 #include <yt/yt/client/table_client/schema.h>
 
-#include <yt/yt/client/chunk_client/read_limit.h>
-
 #include <yt/yt/client/transaction_client/public.h>
 
-#include <yt/yt/client/security_client/public.h>
+#include <yt/yt/core/yson/public.h>
+
+#include <yt/yt/core/ytree/attributes.h>
+
+#include <yt/yt/core/compression/public.h>
+
+#include <yt/yt/library/erasure/public.h>
 
 namespace NYT::NYPath {
 
@@ -167,6 +167,9 @@ class TRichYPath
     // "create"
     bool GetCreate() const;
 
+    // "versioned_read_options"
+    NTableClient::TVersionedReadOptions GetVersionedReadOptions() const;
+
 private:
     TYPath Path_;
     NYTree::IAttributeDictionaryPtr Attributes_;
diff --git a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
@@ -14,6 +14,7 @@ import "yt_proto/yt/core/ytree/proto/request_complexity_limits.proto";
 import "yt_proto/yt/client/chunk_client/proto/data_statistics.proto";
 import "yt_proto/yt/client/chaos_client/proto/replication_card.proto";
 import "yt_proto/yt/client/hive/proto/timestamp_map.proto";
+import "yt_proto/yt/client/table_client/proto/versioned_io_options.proto";
 import "yt_proto/yt/client/tablet_client/proto/lock_mask.proto";
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -608,11 +609,6 @@ message TQueryStatistics
     repeated TQueryStatistics inner_statistics = 14;
 }
 
-message TVersionedReadOptions
-{
-    optional int32 read_mode = 1; // EVersionedIOMode
-}
-
 message TReqSelectRows
 {
     required string query = 1;
@@ -640,7 +636,7 @@ message TReqSelectRows
     optional bool merge_versioned_rows = 21;
     optional int32 syntax_version = 22 [default = 1];
     optional int32 execution_backend = 23; // EExecutionBackend
-    optional TVersionedReadOptions versioned_read_options = 25;
+    optional NYT.NTableClient.NProto.TVersionedReadOptions versioned_read_options = 25;
 
     optional TSuppressableAccessTrackingOptions suppressable_access_tracking_options = 104;
 
diff --git a/yt/yt_proto/yt/client/table_client/proto/versioned_io_options.proto b/yt/yt_proto/yt/client/table_client/proto/versioned_io_options.proto
@@ -0,0 +1,12 @@
+package NYT.NTableClient.NProto;
+
+option go_package = "a.yandex-team.ru/yt/go/proto/client/table_client";
+
+////////////////////////////////////////////////////////////////////////////////
+
+message TVersionedReadOptions
+{
+    required int32 read_mode = 1; // EVersionedIOMode
+}
+
+////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt_proto/yt/client/ya.make b/yt/yt_proto/yt/client/ya.make
@@ -38,6 +38,8 @@ SRCS(
     table_chunk_format/proto/column_meta.proto
     table_chunk_format/proto/wire_protocol.proto
 
+    table_client/proto/versioned_io_options.proto
+
     tablet_client/proto/lock_mask.proto
 
     transaction_client/proto/timestamp_service.proto
