Implement HashV2 and allow to choose it from config (#21391)

Author: abyss7

ydb/core/kqp/compile_service/kqp_compile_actor.cpp

@@ -670,6 +670,15 @@ void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConf
         kqpConfig.YqlCoreOptimizerFlags.insert("fuseequijoinsinputmultilabels");
         kqpConfig.YqlCoreOptimizerFlags.insert("pullupflatmapoverjoinmultiplelabels");
     }
+
+    switch(serviceConfig.GetDefaultHashShuffleFuncType()) {
+        case NKikimrConfig::TTableServiceConfig_EHashKind_HASH_V1:
+            kqpConfig.DefaultHashShuffleFuncType = NYql::NDq::EHashShuffleFuncType::HashV1;
+            break;
+        case NKikimrConfig::TTableServiceConfig_EHashKind_HASH_V2:
+            kqpConfig.DefaultHashShuffleFuncType = NYql::NDq::EHashShuffleFuncType::HashV2;
+            break;
+    }
 }
 
 IActor* CreateKqpCompileActor(const TActorId& owner, const TKqpSettings::TConstPtr& kqpSettings,

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -1844,7 +1844,6 @@ class TKqpExecuterBase : public TActor<TDerived> {
                             Y_ENSURE(hashByShardId.contains(readInfo.ShardId));
                             (*columnShardHashV1Params.TaskIndexByHash)[hashByShardId[readInfo.ShardId]] = stageInternalTaskId;
                         }
-
                     }
                 }
 

ydb/core/kqp/executer_actor/kqp_tasks_graph.cpp

@@ -506,6 +506,10 @@ void BuildKqpStageChannels(TKqpTasksGraph& tasksGraph, TStageInfo& stageInfo,
                         hashKind = EHashShuffleFuncType::HashV1;
                         break;
                     }
+                    case NKqpProto::TKqpPhyCnHashShuffle::kHashV2: {
+                        hashKind = EHashShuffleFuncType::HashV2;
+                        break;
+                    }
                     case NKqpProto::TKqpPhyCnHashShuffle::kColumnShardHashV1: {
                         Y_ENSURE(enableShuffleElimination, "OptShuffleElimination wasn't turned on, but ColumnShardHashV1 detected!");
 
@@ -1083,6 +1087,10 @@ void FillOutputDesc(
                     hashPartitionDesc.MutableHashV1();
                     break;
                 }
+                case HashV2: {
+                    hashPartitionDesc.MutableHashV2();
+                    break;
+                }
                 case ColumnShardHashV1: {
                     auto& columnShardHashV1Params = stageInfo.Meta.GetColumnShardHashV1Params(outputIdx);
                     LOG_DEBUG_S(

ydb/core/kqp/query_compiler/kqp_query_compiler.cpp

@@ -1476,6 +1476,10 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
                     shuffleProto.MutableHashV1();
                     break;
                 }
+                case HashV2: {
+                    shuffleProto.MutableHashV2();
+                    break;
+                }
                 case ColumnShardHashV1: {
                     auto& columnHashV1 = *shuffleProto.MutableColumnShardHashV1();
 

ydb/core/protos/kqp_physical.proto

@@ -252,6 +252,9 @@ message TKqpPhyCnMap {
 message THashV1 {
 }
 
+message THashV2 {
+}
+
 message TColumnShardHashV1 {
     repeated uint32 KeyColumnTypes = 3;
 }
@@ -261,6 +264,7 @@ message TKqpPhyCnHashShuffle {
 
     oneof HashKind {
         THashV1 HashV1 = 2;
+        THashV2 HashV2 = 5;
         TColumnShardHashV1 ColumnShardHashV1 = 3;
     }
 

ydb/core/protos/table_service_config.proto

@@ -396,4 +396,11 @@ message TTableServiceConfig {
     optional bool EnableOlapPushdownProjections = 91  [ default = true ];
 
     optional uint32 DefaultLangVer = 92 [default = 202501];
+
+    enum EHashKind {
+        HASH_V1 = 0;
+        HASH_V2 = 1;
+    }
+
+    optional EHashKind DefaultHashShuffleFuncType = 93 [ default = HASH_V2 ];
 };

ydb/library/yql/dq/common/dq_common.h

@@ -101,6 +101,7 @@ enum class EEnabledSpillingNodes : ui64 {
 
 enum class EHashShuffleFuncType {
     HashV1            = 0     /* "HashV1" */,
+    HashV2            = 2     /* "HashV2" */,
     ColumnShardHashV1 = 1     /* "ColumnShardHashV1" */,
 };
 

ydb/library/yql/dq/proto/dq_tasks.proto

@@ -136,6 +136,9 @@ message TTaskOutputBroadcast {
 message THashV1 {
 }
 
+message THashV2 {
+}
+
 message TColumnShardHashV1 {
     uint64 ShardCount = 1;
     repeated uint64 TaskIndexByHash = 2;
@@ -146,8 +149,9 @@ message TTaskOutputHashPartition {
     repeated string KeyColumns = 1;
     uint32 PartitionsCount = 2;
 
-    oneof THashKind {
+    oneof HashKind {
         THashV1 HashV1 = 3;
+        THashV2 HashV2 = 5;
         TColumnShardHashV1 ColumnShardHashV1 = 4;
     }
 }