[Acc] [PATCH 14/32] uadk: support hashjoin and gather algorithm

7 Aug 2025

From: Wenkai Lin <linwenkai6@hisilicon.com>

UADK supports hardware acceleration for the hashjoin and gather.
Hashjoin is used to construct a hash table to join two tables,
gather is used to combine data of different types in multiple
columns in a specified order to obtain a new column.

Signed-off-by: Wenkai Lin <linwenkai6@hisilicon.com>
Signed-off-by: Qi Tao <taoqi10@huawei.com>
---
 Makefile.am                      |    6 +-
 drv/hisi_dae_join_gather.c       | 1533 +++++++++++++++++++++++++
 include/drv/wd_join_gather_drv.h |   52 +
 include/wd_alg.h                 |    2 +
 include/wd_dae.h                 |   12 +
 include/wd_join_gather.h         |  352 ++++++
 include/wd_util.h                |    1 +
 libwd_dae.map                    |   19 +
 wd_join_gather.c                 | 1821 ++++++++++++++++++++++++++++++
 wd_util.c                        |    6 +-
 10 files changed, 3800 insertions(+), 4 deletions(-)
 create mode 100644 drv/hisi_dae_join_gather.c
 create mode 100644 include/drv/wd_join_gather_drv.h
 create mode 100644 include/wd_join_gather.h
 create mode 100644 wd_join_gather.c

diff --git a/Makefile.am b/Makefile.am
index df756f72..3d312a97 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -37,7 +37,7 @@ pkginclude_HEADERS = include/wd.h include/wd_cipher.h include/wd_aead.h \
 		  include/wd_rsa.h  include/uacce.h include/wd_alg_common.h \
 		  include/wd_ecc.h include/wd_sched.h include/wd_alg.h \
 		  include/wd_zlibwrapper.h include/wd_dae.h include/wd_agg.h \
-		  include/wd_udma.h
+		  include/wd_udma.h include/wd_join_gather.h
 
 nobase_pkginclude_HEADERS = v1/wd.h v1/wd_cipher.h v1/wd_aead.h v1/uacce.h v1/wd_dh.h \
 			 v1/wd_digest.h v1/wd_rsa.h v1/wd_bmm.h
@@ -73,7 +73,7 @@ libwd_la_SOURCES=wd.c wd_mempool.c wd.h	wd_alg.c wd_alg.h	\
 libwd_udma_la_SOURCES=wd_udma.h wd_udma_drv.h wd_udma.c \
 		     wd_util.c wd_util.h wd_sched.c wd_sched.h wd.c wd.h
 
-libwd_dae_la_SOURCES=wd_dae.h wd_agg.h wd_agg_drv.h wd_agg.c \
+libwd_dae_la_SOURCES=wd_dae.h wd_agg.h wd_agg_drv.h wd_agg.c wd_join_gather.h wd_join_gather_drv.h wd_join_gather.c \
 		     wd_util.c wd_util.h wd_sched.c wd_sched.h wd.c wd.h
 
 libwd_comp_la_SOURCES=wd_comp.c wd_comp.h wd_comp_drv.h wd_util.c wd_util.h \
@@ -112,7 +112,7 @@ libisa_sve_la_SOURCES=drv/hash_mb/hash_mb.c wd_digest_drv.h drv/hash_mb/hash_mb.
 endif
 
 libhisi_dae_la_SOURCES=drv/hisi_dae.c drv/hisi_qm_udrv.c \
-		hisi_qm_udrv.h
+		hisi_qm_udrv.h drv/hisi_dae_join_gather.c
 
 libhisi_udma_la_SOURCES=drv/hisi_udma.c drv/hisi_qm_udrv.c \
 		hisi_qm_udrv.h
diff --git a/drv/hisi_dae_join_gather.c b/drv/hisi_dae_join_gather.c
new file mode 100644
index 00000000..da10550b
--- /dev/null
+++ b/drv/hisi_dae_join_gather.c
@@ -0,0 +1,1533 @@
+// SPDX-License-Identifier: Apache-2.0
+/* Copyright 2025 Huawei Technologies Co.,Ltd. All rights reserved. */
+
+#include <math.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
+#include <sys/mman.h>
+#include <sys/types.h>
+#include "hisi_qm_udrv.h"
+#include "../include/drv/wd_join_gather_drv.h"
+
+#define DAE_SQC_ALG_TYPE	2
+#define DAE_EXT_SQE_SIZE	128
+#define DAE_CTX_Q_NUM_DEF	1
+
+/* column information */
+#define DAE_MAX_KEY_COLS	9
+#define DAE_MAX_CHAR_SIZE	32
+#define DAE_MAX_ROW_SIZE	512
+#define DAE_JOIN_MAX_ROW_NUN	50000
+#define DAE_JOIN_MAX_BATCH_NUM	2800
+#define DAE_MAX_TABLE_NUM	16
+
+/* align size */
+#define DAE_CHAR_ALIGN_SIZE	4
+#define DAE_KEY_ALIGN_SIZE	8
+#define DAE_TABLE_ALIGN_SIZE	128
+#define DAE_ADDR_ALIGN_SIZE	128
+#define DAE_BREAKPOINT_SIZE	81920
+#define DAE_ADDR_INDEX_SHIFT	1
+
+/* hash table */
+#define HASH_EXT_TABLE_INVALID_OFFSET	5
+#define HASH_EXT_TABLE_VALID	0x80
+#define HASH_TABLE_HEAD_TAIL_SIZE	8
+#define HASH_TABLE_INDEX_NUM		1
+#define HASH_TABLE_MAX_INDEX_NUM	15
+#define HASH_TABLE_INDEX_SIZE		12
+#define HASH_TABLE_EMPTY_SIZE	4
+#define GATHER_ROW_BATCH_EMPTY_SIZE	2
+#define HASH_TABLE_WITDH_POWER		2
+#define HASH_TABLE_MIN_WIDTH	10
+#define HASH_TABLE_MAX_WIDTH	43
+#define HASH_TABLE_OFFSET_3ROW		3
+#define HASH_TABLE_OFFSET_1ROW		1
+
+#define __ALIGN_MASK(x, mask)  (((x) + (mask)) & ~(mask))
+#define ALIGN(x, a) __ALIGN_MASK(x, (typeof(x))(a)-1)
+#define PTR_ALIGN(p, a)	((typeof(p))ALIGN((uintptr_t)(p), (a)))
+
+#define BIT(nr)		(1UL << (nr))
+#define BITS_PER_LONG	(__SIZEOF_LONG__ * 8)
+#define GENMASK(h, l) \
+	(((~0UL) << (l)) & (~0UL >> (BITS_PER_LONG - 1 - (h))))
+
+/* DAE hardware protocol data */
+enum dae_join_stage {
+	DAE_JOIN_BUILD_HASH = 0x0,
+	DAE_JOIN_REHASH = 0x6,
+	DAE_JOIN_PROBE = 0x7,
+};
+
+enum dae_gather_stage {
+	DAE_GATHER_CONVERT = 0x0,
+	DAE_GATHER_COMPLETE = 0x7,
+};
+
+enum dae_task_type {
+	DAE_HASH_JOIN = 0x1,
+	DAE_GATHER = 0x2,
+};
+
+enum dae_done_flag {
+	DAE_HW_TASK_NOT_PROCESS = 0x0,
+	DAE_HW_TASK_DONE = 0x1,
+	DAE_HW_TASK_ERR = 0x2,
+};
+
+enum dae_error_type {
+	DAE_TASK_SUCCESS = 0x0,
+	DAE_TASK_BD_ERROR_MIN = 0x1,
+	DAE_TASK_BD_ERROR_MAX = 0x7f,
+	DAE_HASH_TABLE_NEED_REHASH = 0x82,
+	DAE_HASH_TABLE_INVALID = 0x83,
+	DAE_JOIN_GATHER_BUS_ERROR = 0x86,
+};
+
+enum dae_data_type {
+	DAE_SINT32 = 0x0,
+	DAE_SINT64 = 0x2,
+	DAE_DECIMAL64 = 0x9,
+	DAE_DECIMAL128 = 0xA,
+	DAE_CHAR = 0xC,
+	DAE_VCHAR = 0xD,
+};
+
+enum dae_date_type_size {
+	SINT32_SIZE = 4,
+	SINT64_SIZE = 8,
+	DECIMAL128_SIZE = 16,
+	DEFAULT_VCHAR_SIZE = 30,
+};
+
+enum dae_table_row_size {
+	ROW_SIZE32 = 32,
+	ROW_SIZE64 = 64,
+	ROW_SIZE128 = 128,
+	ROW_SIZE256 = 256,
+	ROW_SIZE512 = 512,
+};
+
+enum dae_bd_type {
+	DAE_BD_TYPE_V2 = 0x1,
+};
+
+struct dae_sqe {
+	__u32 bd_type : 6;
+	__u32 resv1 : 2;
+	__u32 task_type : 6;
+	__u32 resv2 : 2;
+	__u32 task_type_ext : 6;
+	__u32 resv3 : 9;
+	__u32 bd_invlid : 1;
+	__u16 table_row_size;
+	__u16 resv4;
+	__u32 batch_num;
+	__u32 low_tag;
+	__u32 hi_tag;
+	__u32 data_row_num;
+	__u32 init_row_num;
+	__u32 src_table_width : 6;
+	__u32 dst_table_width : 6;
+	__u32 key_out_en : 1;
+	__u32 break_point_en : 1;
+	__u32 multi_batch_en : 1;
+	__u32 sva_prefetch_en : 1;
+	__u32 counta_vld : 1;
+	__u32 index_num : 5;
+	__u32 resv5 : 8;
+	__u32 index_batch_type : 1;
+	__u32 resv6 : 1;
+	/*
+	 * high 4bits: compare mode if data type is char/vchar,
+	 *             out type if operation is sum.
+	 * low 4bits: input value type.
+	 */
+	__u8 key_data_type[16];
+	__u32 resv7[10];
+	__u64 addr_ext;
+	__u16 key_col_bitmap;
+	__u16 has_empty;
+	__u32 resv8;
+	__u64 addr_list;
+	__u32 done_flag : 3;
+	__u32 output_end : 1;
+	__u32 ext_err_type : 12;
+	__u32 err_type : 8;
+	__u32 wtype : 8;
+	__u32 out_raw_num;
+	__u32 next_init_row_num;
+	__u32 resv9;
+};
+
+struct dae_ext_sqe {
+	/*
+	 * If date type is char/vchar, data info fill data type size
+	 * If data type is decimal64/decimal128, data info fill data precision
+	 */
+	__u16 key_data_info[16];
+	__u16 resv1[16];
+	__u64 resv2[2];
+	__u32 resv3[12];
+};
+
+struct dae_col_addr {
+	__u64 empty_addr;
+	__u64 empty_size;
+	__u64 value_addr;
+	__u64 value_size;
+};
+
+struct dae_table_addr {
+	__u64 std_table_addr;
+	__u64 std_table_size;
+	__u64 ext_table_addr;
+	__u64 ext_table_size;
+};
+
+struct dae_probe_info_addr {
+	__u64 batch_num_index;
+	__u64 batch_addr_index;
+	__u64 probe_index_addr;
+	__u64 resv1;
+	__u64 break_point_addr;
+	__u64 resv2;
+};
+
+struct dae_addr_list {
+	__u64 ext_sqe_addr;
+	__u64 ext_sqe_size;
+	struct dae_table_addr src_table;
+	struct dae_table_addr dst_table;
+	struct dae_probe_info_addr probe_info;
+	struct dae_col_addr input_addr[32];
+	struct dae_col_addr output_addr[32];
+};
+
+struct dae_extend_addr {
+	struct dae_ext_sqe *ext_sqe;
+	struct dae_addr_list *addr_list;
+	__u8 *addr_status;
+	__u16 addr_num;
+	__u16 tail;
+};
+
+static enum dae_data_type hw_data_type_order[] = {
+	DAE_VCHAR, DAE_CHAR, DAE_DECIMAL128,
+	DAE_DECIMAL64, DAE_SINT64, DAE_SINT32,
+};
+
+struct hw_join_gather_data {
+	enum dae_data_type hw_type;
+	__u32 optype;
+	__u32 usr_col_idx;
+	__u16 data_info;
+};
+
+struct join_gather_col_data {
+	struct hw_join_gather_data key_data[DAE_MAX_KEY_COLS];
+	struct hw_join_gather_data gather_data[DAE_MAX_TABLE_NUM][DAE_MAX_KEY_COLS];
+
+	__u32 key_num;
+	__u32 gather_table_num;
+	__u32 gather_cols_num[DAE_MAX_TABLE_NUM];
+	__u16 has_empty[DAE_MAX_TABLE_NUM];
+	__u8 index_num;
+};
+
+struct hash_table_data {
+	void *std_table;
+	void *ext_table;
+	__u64 std_table_size;
+	__u64 ext_table_size;
+	__u32 table_width;
+};
+
+struct join_gather_ctx {
+	struct join_gather_col_data cols_data;
+	struct hash_table_data table_data;
+	struct hash_table_data rehash_table;
+	pthread_spinlock_t lock;
+	__u32 hash_table_row_size;
+	__u32 batch_row_size[DAE_MAX_TABLE_NUM];
+};
+
+struct hisi_dae_ctx {
+	struct wd_ctx_config_internal config;
+};
+
+static int get_free_ext_addr(struct dae_extend_addr *ext_addr)
+{
+	__u16 addr_num = ext_addr->addr_num;
+	__u16 idx = ext_addr->tail;
+	__u16 cnt = 0;
+
+	/* The addr_num is equal to sq_depth(1024). */
+	while (__atomic_test_and_set(&ext_addr->addr_status[idx], __ATOMIC_ACQUIRE)) {
+		idx = (idx + 1) % addr_num;
+		cnt++;
+		if (cnt == addr_num)
+			return -WD_EBUSY;
+	}
+
+	ext_addr->tail = (idx + 1) % addr_num;
+
+	return idx;
+}
+
+static void put_ext_addr(struct dae_extend_addr *ext_addr, int idx)
+{
+	__atomic_clear(&ext_addr->addr_status[idx], __ATOMIC_RELEASE);
+}
+
+static void fill_join_gather_misc_field(struct wd_join_gather_msg *msg,
+					struct dae_sqe *sqe)
+{
+	struct join_gather_ctx *ctx = msg->priv;
+	struct join_gather_col_data *cols_data = &ctx->cols_data;
+
+	sqe->sva_prefetch_en = true;
+
+	switch (msg->op_type) {
+	case WD_JOIN_BUILD_HASH:
+		sqe->task_type = DAE_HASH_JOIN;
+		sqe->task_type_ext = DAE_JOIN_BUILD_HASH;
+		sqe->data_row_num = msg->req.input_row_num;
+		sqe->batch_num = msg->req.join_req.build_batch_index;
+		sqe->init_row_num = msg->req.join_req.batch_row_offset;
+		sqe->index_num = cols_data->index_num;
+		break;
+	case WD_JOIN_PROBE:
+		sqe->task_type = DAE_HASH_JOIN;
+		sqe->task_type_ext = DAE_JOIN_PROBE;
+		sqe->data_row_num = msg->req.output_row_num;
+		sqe->batch_num = msg->req.input_row_num;
+		sqe->init_row_num = msg->req.join_req.batch_row_offset;
+		sqe->index_num = cols_data->index_num;
+		sqe->key_out_en = msg->key_out_en;
+		sqe->break_point_en = sqe->init_row_num ? true : false;
+		sqe->index_batch_type = msg->index_type;
+		break;
+	case WD_JOIN_REHASH:
+		sqe->task_type = DAE_HASH_JOIN;
+		sqe->task_type_ext = DAE_JOIN_REHASH;
+		sqe->data_row_num = msg->req.output_row_num;
+		sqe->index_num = cols_data->index_num;
+		break;
+	case WD_GATHER_CONVERT:
+		sqe->task_type = DAE_GATHER;
+		sqe->task_type_ext = DAE_GATHER_CONVERT;
+		sqe->data_row_num = msg->req.input_row_num;
+		break;
+	case WD_GATHER_COMPLETE:
+		sqe->task_type = DAE_GATHER;
+		sqe->task_type_ext = DAE_GATHER_COMPLETE;
+		sqe->multi_batch_en = msg->multi_batch_en;
+		sqe->index_batch_type = msg->index_type;
+		sqe->data_row_num = msg->req.output_row_num;
+		break;
+	default:
+		break;
+	}
+}
+
+static void fill_join_table_data(struct dae_sqe *sqe, struct dae_addr_list *addr_list,
+				 struct wd_join_gather_msg *msg)
+{
+	struct join_gather_ctx *ctx = (struct join_gather_ctx *)msg->priv;
+	struct dae_table_addr *hw_table_src = &addr_list->src_table;
+	struct dae_table_addr *hw_table_dst = &addr_list->dst_table;
+	struct hash_table_data *table_data_src, *table_data_dst;
+
+	switch (msg->op_type) {
+	case WD_JOIN_BUILD_HASH:
+		table_data_src = NULL;
+		table_data_dst = &ctx->table_data;
+		break;
+	case WD_JOIN_REHASH:
+		table_data_src = &ctx->rehash_table;
+		table_data_dst = &ctx->table_data;
+		break;
+	case WD_JOIN_PROBE:
+		table_data_src = &ctx->table_data;
+		table_data_dst = NULL;
+		break;
+	default:
+		return;
+	}
+
+	sqe->table_row_size = ctx->hash_table_row_size;
+
+	if (table_data_src) {
+		sqe->src_table_width = table_data_src->table_width;
+		hw_table_src->std_table_addr = (__u64)(uintptr_t)table_data_src->std_table;
+		hw_table_src->std_table_size = table_data_src->std_table_size;
+		hw_table_src->ext_table_addr = (__u64)(uintptr_t)table_data_src->ext_table;
+		hw_table_src->ext_table_size = table_data_src->ext_table_size;
+	}
+
+	if (table_data_dst) {
+		sqe->dst_table_width = table_data_dst->table_width;
+		hw_table_dst->std_table_addr = (__u64)(uintptr_t)table_data_dst->std_table;
+		hw_table_dst->std_table_size = table_data_dst->std_table_size;
+		hw_table_dst->ext_table_addr = (__u64)(uintptr_t)table_data_dst->ext_table;
+		hw_table_dst->ext_table_size = table_data_dst->ext_table_size;
+	}
+}
+
+static void fill_join_key_data(struct dae_sqe *sqe, struct dae_ext_sqe *ext_sqe,
+			       struct dae_addr_list *addr_list,
+			       struct wd_join_gather_msg *msg)
+{
+	struct dae_probe_info_addr *info = &addr_list->probe_info;
+	struct join_gather_ctx *ctx = msg->priv;
+	struct hw_join_gather_data *key_data = ctx->cols_data.key_data;
+	struct wd_dae_col_addr *usr_key, *out_usr_key = NULL;
+	struct dae_col_addr *hw_key, *out_hw_key = NULL;
+	struct wd_join_req *req = &msg->req.join_req;
+	struct wd_probe_out_info *output = &req->probe_output;
+	__u16 usr_col_idx;
+	__u64 offset;
+	__u32 i;
+
+	sqe->key_col_bitmap = GENMASK(msg->key_cols_num - 1, 0);
+
+	for (i = 0; i < msg->key_cols_num; i++) {
+		sqe->key_data_type[i] = key_data[i].hw_type;
+		ext_sqe->key_data_info[i] = key_data[i].data_info;
+	}
+
+	switch (msg->op_type) {
+	case WD_JOIN_BUILD_HASH:
+		usr_key = req->key_cols;
+		hw_key = addr_list->input_addr;
+		if (msg->index_type == WD_BATCH_ADDR_INDEX)
+			sqe->addr_ext = (__u64)(uintptr_t)req->build_batch_addr.addr;
+		break;
+	case WD_JOIN_PROBE:
+		usr_key = req->key_cols;
+		hw_key = addr_list->input_addr;
+		if (msg->key_out_en) {
+			out_usr_key = output->key_cols;
+			out_hw_key = addr_list->output_addr;
+		}
+
+		info->batch_num_index = (__u64)(uintptr_t)output->build_index.addr;
+		info->probe_index_addr = (__u64)(uintptr_t)output->probe_index.addr;
+		info->break_point_addr = (__u64)(uintptr_t)output->breakpoint.addr;
+
+		if (msg->index_type == WD_BATCH_ADDR_INDEX) {
+			offset = (__u64)output->build_index.row_size * output->build_index.row_num;
+			offset = offset >> DAE_ADDR_INDEX_SHIFT;
+			info->batch_addr_index = info->batch_num_index + offset;
+		}
+		break;
+	default:
+		return;
+	}
+
+	for (i = 0; i < msg->key_cols_num; i++) {
+		usr_col_idx = key_data[i].usr_col_idx;
+		hw_key[i].empty_addr = (__u64)(uintptr_t)usr_key[usr_col_idx].empty;
+		hw_key[i].empty_size = usr_key[usr_col_idx].empty_size;
+		hw_key[i].value_addr = (__u64)(uintptr_t)usr_key[usr_col_idx].value;
+		hw_key[i].value_size = usr_key[usr_col_idx].value_size;
+
+		if (!out_usr_key)
+			continue;
+		out_hw_key[i].empty_addr = (__u64)(uintptr_t)out_usr_key[usr_col_idx].empty;
+		out_hw_key[i].empty_size = out_usr_key[usr_col_idx].empty_size;
+		/* The hardware does not output the empty data, set the data by software. */
+		memset(out_usr_key[usr_col_idx].empty, 0, out_usr_key[usr_col_idx].empty_size);
+
+		out_hw_key[i].value_addr = (__u64)(uintptr_t)out_usr_key[usr_col_idx].value;
+		out_hw_key[i].value_size = out_usr_key[usr_col_idx].value_size;
+	}
+}
+
+static void fill_gather_col_data(struct dae_sqe *sqe, struct dae_ext_sqe *ext_sqe,
+				 struct dae_addr_list *addr_list,
+				 struct wd_join_gather_msg *msg)
+{
+	struct dae_probe_info_addr *info = &addr_list->probe_info;
+	struct join_gather_ctx *ctx = msg->priv;
+	struct join_gather_col_data *cols_data = &ctx->cols_data;
+	struct wd_gather_req *gather_req = &msg->req.gather_req;
+	__u32 table_index = gather_req->table_index;
+	struct hw_join_gather_data *gather_data = cols_data->gather_data[table_index];
+	__u16 cols_num = cols_data->gather_cols_num[table_index];
+	struct wd_dae_col_addr *usr_data;
+	struct dae_col_addr *hw_data;
+	__u16 usr_col_idx;
+	void **batch_addr;
+	__u64 offset;
+	__u32 i;
+
+	sqe->key_col_bitmap = GENMASK(cols_num - 1, 0);
+	sqe->has_empty = cols_data->has_empty[table_index];
+	sqe->table_row_size = ctx->batch_row_size[table_index];
+
+	usr_data = gather_req->data_cols;
+	batch_addr = gather_req->row_batchs.batch_addr;
+
+	switch (msg->op_type) {
+	case WD_GATHER_CONVERT:
+		hw_data = addr_list->input_addr;
+		/* Single batch tasks use the first element of the array. */
+		addr_list->dst_table.std_table_addr = (__u64)(uintptr_t)batch_addr[0];
+		break;
+	case WD_GATHER_COMPLETE:
+		hw_data = addr_list->output_addr;
+		if (!msg->multi_batch_en) {
+			info->probe_index_addr = (__u64)(uintptr_t)gather_req->index.addr;
+			addr_list->src_table.std_table_addr = (__u64)(uintptr_t)batch_addr[0];
+			break;
+		}
+
+		info->batch_num_index = (__u64)(uintptr_t)gather_req->index.addr;
+		if (msg->index_type == WD_BATCH_ADDR_INDEX) {
+			offset = (__u64)gather_req->index.row_size * gather_req->index.row_num;
+			offset = offset >> DAE_ADDR_INDEX_SHIFT;
+			info->batch_addr_index = info->batch_num_index + offset;
+		} else {
+			addr_list->src_table.std_table_addr = (__u64)(uintptr_t)batch_addr;
+		}
+		break;
+	default:
+		return;
+	}
+
+	for (i = 0; i < cols_num; i++) {
+		sqe->key_data_type[i] = gather_data[i].hw_type;
+		ext_sqe->key_data_info[i] = gather_data[i].data_info;
+
+		usr_col_idx = gather_data[i].usr_col_idx;
+		hw_data[i].empty_addr = (__u64)(uintptr_t)usr_data[usr_col_idx].empty;
+		hw_data[i].empty_size = usr_data[usr_col_idx].empty_size;
+		hw_data[i].value_addr = (__u64)(uintptr_t)usr_data[usr_col_idx].value;
+		hw_data[i].value_size = usr_data[usr_col_idx].value_size;
+	}
+}
+
+static void fill_join_gather_ext_addr(struct dae_sqe *sqe, struct dae_ext_sqe *ext_sqe,
+				      struct dae_addr_list *addr_list)
+{
+	memset(ext_sqe, 0, DAE_EXT_SQE_SIZE);
+	memset(addr_list, 0, sizeof(struct dae_addr_list));
+	sqe->addr_list = (__u64)(uintptr_t)addr_list;
+	addr_list->ext_sqe_addr = (__u64)(uintptr_t)ext_sqe;
+	addr_list->ext_sqe_size = DAE_EXT_SQE_SIZE;
+}
+
+static void fill_join_gather_info(struct dae_sqe *sqe, struct dae_ext_sqe *ext_sqe,
+				  struct dae_addr_list *addr_list,
+				  struct wd_join_gather_msg *msg)
+{
+	fill_join_gather_ext_addr(sqe, ext_sqe, addr_list);
+	sqe->bd_type = DAE_BD_TYPE_V2;
+
+	switch (msg->op_type) {
+	case WD_JOIN_BUILD_HASH:
+	case WD_JOIN_PROBE:
+	case WD_JOIN_REHASH:
+		fill_join_table_data(sqe, addr_list, msg);
+		fill_join_key_data(sqe, ext_sqe, addr_list, msg);
+		break;
+	case WD_GATHER_CONVERT:
+	case WD_GATHER_COMPLETE:
+		fill_gather_col_data(sqe, ext_sqe, addr_list, msg);
+		break;
+	default:
+		break;
+	}
+}
+
+static int check_join_gather_param(struct wd_join_gather_msg *msg)
+{
+	struct wd_probe_out_info *output = &msg->req.join_req.probe_output;
+	__u64 row_num;
+	__u64 size;
+
+	if (!msg) {
+		WD_ERR("invalid: input join gather msg is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	switch (msg->op_type) {
+	case WD_JOIN_BUILD_HASH:
+		if (msg->req.input_row_num > DAE_JOIN_MAX_ROW_NUN) {
+			WD_ERR("invalid: build table row count %u is more than %d!\n",
+			       msg->req.input_row_num, DAE_JOIN_MAX_ROW_NUN);
+			return -WD_EINVAL;
+		}
+		if (msg->index_type == WD_BATCH_NUMBER_INDEX) {
+			if (msg->req.join_req.build_batch_index >= DAE_JOIN_MAX_BATCH_NUM) {
+				WD_ERR("invalid: input join batch index is more than %d!\n",
+				       DAE_JOIN_MAX_BATCH_NUM - 1);
+				return -WD_EINVAL;
+			}
+		} else {
+			if (!msg->req.join_req.build_batch_addr.addr ||
+			    !msg->req.join_req.build_batch_addr.row_num ||
+			    !msg->req.join_req.build_batch_addr.row_size) {
+				WD_ERR("invalid: input join build batch addr is NULL!\n");
+				return -WD_EINVAL;
+			}
+		}
+		break;
+	case WD_JOIN_PROBE:
+		size = (__u64)output->breakpoint.row_size * output->breakpoint.row_num;
+		if (!output->breakpoint.addr || size < DAE_BREAKPOINT_SIZE) {
+			WD_ERR("invalid probe breakpoint size: %llu\n", size);
+			return -WD_EINVAL;
+		}
+		if (msg->index_type == WD_BATCH_ADDR_INDEX) {
+			row_num = output->probe_index.row_num << DAE_ADDR_INDEX_SHIFT;
+			if (output->build_index.row_num != row_num) {
+				WD_ERR("invalid: build index row number need be %llu\n", row_num);
+				return -WD_EINVAL;
+			}
+		}
+		break;
+	case WD_JOIN_REHASH:
+	case WD_GATHER_CONVERT:
+		break;
+	case WD_GATHER_COMPLETE:
+		if (msg->index_type == WD_BATCH_NUMBER_INDEX) {
+			if (msg->req.gather_req.row_batchs.batch_num > DAE_JOIN_MAX_BATCH_NUM) {
+				WD_ERR("invalid: gather row batch num is more than %d!\n",
+				       DAE_JOIN_MAX_BATCH_NUM);
+				return -WD_EINVAL;
+			}
+		} else {
+			row_num = output->probe_index.row_num << DAE_ADDR_INDEX_SHIFT;
+			if (output->build_index.row_num != row_num) {
+				WD_ERR("invalid: build index row number need be: %llu\n", row_num);
+				return -WD_EINVAL;
+			}
+		}
+		break;
+	default:
+		break;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int join_gather_send(struct wd_alg_driver *drv, handle_t ctx, void *send_msg)
+{
+	handle_t h_qp = (handle_t)wd_ctx_get_priv(ctx);
+	struct hisi_qp *qp = (struct hisi_qp *)h_qp;
+	struct dae_extend_addr *ext_addr = qp->priv;
+	struct wd_join_gather_msg *msg = send_msg;
+	struct dae_addr_list *addr_list;
+	struct dae_ext_sqe *ext_sqe;
+	struct dae_sqe sqe = {0};
+	__u16 send_cnt = 0;
+	int ret, idx;
+
+	ret = check_join_gather_param(msg);
+	if (ret)
+		return ret;
+
+	fill_join_gather_misc_field(msg, &sqe);
+
+	idx = get_free_ext_addr(ext_addr);
+	if (idx < 0)
+		return -WD_EBUSY;
+	addr_list = &ext_addr->addr_list[idx];
+	ext_sqe = &ext_addr->ext_sqe[idx];
+
+	fill_join_gather_info(&sqe, ext_sqe, addr_list, msg);
+
+	hisi_set_msg_id(h_qp, &msg->tag);
+	sqe.low_tag = msg->tag;
+	sqe.hi_tag = idx;
+
+	ret = hisi_qm_send(h_qp, &sqe, 1, &send_cnt);
+	if (ret) {
+		if (ret != -WD_EBUSY)
+			WD_ERR("failed to send to hardware, ret = %d!\n", ret);
+		put_ext_addr(ext_addr, idx);
+		return ret;
+	}
+
+	return WD_SUCCESS;
+}
+
+static void fill_join_gather_task_done(struct dae_sqe *sqe, struct wd_join_gather_msg *msg,
+				       struct wd_join_gather_msg *temp_msg,
+				       struct join_gather_ctx *ctx)
+{
+	if (sqe->task_type == DAE_HASH_JOIN) {
+		if (sqe->task_type_ext == DAE_JOIN_PROBE) {
+			msg->consumed_row_num = sqe->next_init_row_num;
+			msg->produced_row_num = sqe->out_raw_num;
+			msg->output_done = sqe->output_end;
+		} else if (sqe->task_type_ext == DAE_JOIN_REHASH) {
+			msg->output_done = sqe->output_end;
+		}
+	}
+}
+
+static void fill_join_gather_task_err(struct dae_sqe *sqe, struct wd_join_gather_msg *msg,
+				      struct wd_join_gather_msg *temp_msg,
+				      struct join_gather_ctx *ctx)
+{
+	switch (sqe->err_type) {
+	case DAE_TASK_BD_ERROR_MIN ... DAE_TASK_BD_ERROR_MAX:
+		WD_ERR("failed to do join gather task, bd error=0x%x!\n", sqe->err_type);
+		msg->result = WD_JOIN_GATHER_PARSE_ERROR;
+		break;
+	case DAE_HASH_TABLE_NEED_REHASH:
+		msg->result = WD_JOIN_GATHER_NEED_REHASH;
+		break;
+	case DAE_HASH_TABLE_INVALID:
+		msg->result = WD_JOIN_GATHER_INVALID_HASH_TABLE;
+		break;
+	case DAE_JOIN_GATHER_BUS_ERROR:
+		WD_ERR("failed to do join gather task, bus error %u!\n", sqe->err_type);
+		msg->result = WD_JOIN_GATHER_BUS_ERROR;
+		break;
+	default:
+		WD_ERR("failed to do dae task! done_flag=0x%x, etype=0x%x, ext_type = 0x%x!\n",
+			(__u32)sqe->done_flag, (__u32)sqe->err_type, (__u32)sqe->ext_err_type);
+		msg->result = WD_JOIN_GATHER_PARSE_ERROR;
+		break;
+	}
+
+	if (sqe->task_type == DAE_HASH_JOIN && sqe->task_type_ext == DAE_JOIN_PROBE) {
+		msg->produced_row_num = sqe->out_raw_num;
+		msg->consumed_row_num = sqe->next_init_row_num;
+		msg->output_done = sqe->output_end;
+	}
+}
+
+static int join_gather_recv(struct wd_alg_driver *drv, handle_t hctx, void *recv_msg)
+{
+	handle_t h_qp = (handle_t)wd_ctx_get_priv(hctx);
+	struct hisi_qp *qp = (struct hisi_qp *)h_qp;
+	struct dae_extend_addr *ext_addr = qp->priv;
+	struct wd_join_gather_msg *msg = recv_msg;
+	struct wd_join_gather_msg *temp_msg = msg;
+	struct join_gather_ctx *ctx;
+	struct dae_sqe sqe = {0};
+	__u16 recv_cnt = 0;
+	int ret;
+
+	ret = hisi_qm_recv(h_qp, &sqe, 1, &recv_cnt);
+	if (ret)
+		return ret;
+
+	ret = hisi_check_bd_id(h_qp, msg->tag, sqe.low_tag);
+	if (ret)
+		goto out;
+
+	msg->tag = sqe.low_tag;
+	if (qp->q_info.qp_mode == CTX_MODE_ASYNC) {
+		temp_msg = wd_join_gather_get_msg(qp->q_info.idx, msg->tag);
+		if (!temp_msg) {
+			msg->result = WD_JOIN_GATHER_IN_EPARA;
+			WD_ERR("failed to get send msg! idx = %u, tag = %u.\n",
+			       qp->q_info.idx, msg->tag);
+			ret = -WD_EINVAL;
+			goto out;
+		}
+	}
+
+	ctx = (struct join_gather_ctx *)temp_msg->priv;
+	msg->result = WD_JOIN_GATHER_TASK_DONE;
+	msg->consumed_row_num = 0;
+
+	if (likely(sqe.done_flag == DAE_HW_TASK_DONE)) {
+		fill_join_gather_task_done(&sqe, msg, temp_msg, ctx);
+	} else if (sqe.done_flag == DAE_HW_TASK_ERR) {
+		fill_join_gather_task_err(&sqe, msg, temp_msg, ctx);
+	} else {
+		msg->result = WD_JOIN_GATHER_PARSE_ERROR;
+		WD_ERR("failed to do join gather task, hardware doesn't process the task!\n");
+	}
+
+out:
+	put_ext_addr(ext_addr, sqe.hi_tag);
+	return ret;
+}
+
+static int join_check_params(struct wd_join_gather_col_info *key_info, __u32 cols_num)
+{
+	__u32 i;
+
+	if (cols_num > DAE_MAX_KEY_COLS) {
+		WD_ERR("invalid: join key cols num %u is more than device support %d!\n",
+			cols_num, DAE_MAX_KEY_COLS);
+		return -WD_EINVAL;
+	}
+
+	for (i = 0; i < cols_num; i++) {
+		switch (key_info[i].data_type) {
+		case WD_DAE_CHAR:
+		case WD_DAE_VARCHAR:
+			WD_ERR("invalid: key col %u, char or varchar isn't supported!\n", i);
+			return -WD_EINVAL;
+		default:
+			break;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int gather_check_params(struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_gather_table_info *table = setup->gather_tables;
+	struct wd_join_gather_col_info *col;
+	__u32 i, j;
+
+	if (setup->gather_table_num > DAE_MAX_TABLE_NUM) {
+		WD_ERR("invalid: gather table num %u is more than device support %d!\n",
+		       setup->gather_table_num, DAE_MAX_TABLE_NUM);
+		return -WD_EINVAL;
+	}
+
+	for (i = 0; i < setup->gather_table_num; i++) {
+		col = table[i].cols;
+		if (table[i].cols_num > DAE_MAX_KEY_COLS) {
+			WD_ERR("invalid: gather cols num %u is more than device support %d!\n",
+			       table[i].cols_num, DAE_MAX_KEY_COLS);
+			return -WD_EINVAL;
+		}
+		for (j = 0; j < table[i].cols_num; i++) {
+			switch (col[j].data_type) {
+			case WD_DAE_CHAR:
+				if (col[j].data_info > DAE_MAX_CHAR_SIZE) {
+					WD_ERR("gather col %u, char size isn't supported!\n", i);
+					return -WD_EINVAL;
+				}
+				break;
+			case WD_DAE_VARCHAR:
+				WD_ERR("invalid: gather col %u, varchar isn't supported!\n", i);
+				return -WD_EINVAL;
+			default:
+				break;
+			}
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int join_gather_param_check(struct wd_join_gather_sess_setup *setup, __u16 hw_type)
+{
+	int ret;
+
+	switch (setup->alg) {
+	case WD_JOIN:
+		return join_check_params(setup->join_table.build_key_cols,
+					 setup->join_table.build_key_cols_num);
+	case WD_GATHER:
+		return gather_check_params(setup);
+	case WD_JOIN_GATHER:
+		ret = join_check_params(setup->join_table.build_key_cols,
+					setup->join_table.build_key_cols_num);
+		if (ret)
+			return ret;
+
+		return gather_check_params(setup);
+	default:
+		return -WD_EINVAL;
+	}
+}
+
+static __u32 get_data_type_size(enum dae_data_type type, __u16 data_info)
+{
+	switch (type) {
+	case DAE_SINT32:
+		return SINT32_SIZE;
+	case DAE_SINT64:
+	case DAE_DECIMAL64:
+		return SINT64_SIZE;
+	case DAE_DECIMAL128:
+		return DECIMAL128_SIZE;
+	case DAE_CHAR:
+		return ALIGN(data_info, DAE_CHAR_ALIGN_SIZE);
+	default:
+		break;
+	}
+
+	return 0;
+}
+
+static int transfer_col_info(struct wd_join_gather_col_info *cols,
+			     struct hw_join_gather_data *data, __u32 col_num)
+{
+	__u32 i;
+
+	for (i = 0; i < col_num; i++) {
+		switch (cols[i].data_type) {
+		case WD_DAE_CHAR:
+			data[i].hw_type = DAE_CHAR;
+			data[i].data_info = cols[i].data_info;
+			break;
+		case WD_DAE_LONG_DECIMAL:
+			data[i].hw_type = DAE_DECIMAL128;
+			break;
+		case WD_DAE_SHORT_DECIMAL:
+			data[i].hw_type = DAE_DECIMAL64;
+			break;
+		case WD_DAE_LONG:
+			data[i].hw_type = DAE_SINT64;
+			break;
+		case WD_DAE_INT:
+		case WD_DAE_DATE:
+			data[i].hw_type = DAE_SINT32;
+			break;
+		default:
+			return -WD_EINVAL;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int transfer_data_to_hw_type(struct join_gather_col_data *cols_data,
+				    struct wd_join_gather_sess_setup *setup)
+{
+	struct hw_join_gather_data tmp_data[DAE_MAX_KEY_COLS] = {0};
+	struct wd_gather_table_info *tables = setup->gather_tables;
+	struct wd_join_gather_col_info *gather_cols;
+	struct hw_join_gather_data *hw_data;
+	__u32 type_num = ARRAY_SIZE(hw_data_type_order);
+	__u32 n, i, j, k = 0;
+	int ret;
+
+	for (n = 0; n < setup->gather_table_num; n++) {
+		gather_cols = tables[n].cols;
+		hw_data = cols_data->gather_data[n];
+		/* Set k to 0 for each table. */
+		k = 0;
+		ret = transfer_col_info(gather_cols, tmp_data, tables[n].cols_num);
+		if (ret)
+			return ret;
+
+		for (i = 0; i < type_num; i++) {
+			for (j = 0; j < tables[n].cols_num; j++) {
+				if (hw_data_type_order[i] != tmp_data[j].hw_type)
+					continue;
+				hw_data[k].usr_col_idx = j;
+				hw_data[k].hw_type = tmp_data[j].hw_type;
+				hw_data[k++].data_info = tmp_data[j].data_info;
+			}
+		}
+
+		cols_data->gather_cols_num[n] = tables[n].cols_num;
+		for (j = 0; j < tables[n].cols_num; j++)
+			if (gather_cols[j].has_empty)
+				cols_data->has_empty[n] |= (1 << j);
+	}
+
+	return WD_SUCCESS;
+}
+
+static int transfer_key_to_hw_type(struct join_gather_col_data *cols_data,
+				   struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_join_gather_col_info *key_cols = setup->join_table.build_key_cols;
+	struct hw_join_gather_data tmp_key_data[DAE_MAX_KEY_COLS] = {0};
+	struct hw_join_gather_data *hw_key_data = cols_data->key_data;
+	__u32 cols_num = setup->join_table.build_key_cols_num;
+	__u32 type_num = ARRAY_SIZE(hw_data_type_order);
+	__u32 i, j, k = 0;
+	int ret;
+
+	ret = transfer_col_info(key_cols, tmp_key_data, cols_num);
+	if (ret)
+		return ret;
+
+	for (i = 0; i < type_num; i++) {
+		for (j = 0; j < cols_num; j++) {
+			if (hw_data_type_order[i] != tmp_key_data[j].hw_type)
+				continue;
+			hw_key_data[k].usr_col_idx = j;
+			hw_key_data[k].hw_type = tmp_key_data[j].hw_type;
+			hw_key_data[k++].data_info = tmp_key_data[j].data_info;
+		}
+	}
+
+	cols_data->key_num = cols_num;
+
+	return WD_SUCCESS;
+}
+
+static int join_get_table_rowsize(struct join_gather_col_data *cols_data,
+				  struct wd_join_gather_sess_setup *setup)
+{
+	struct hw_join_gather_data *key_data = cols_data->key_data;
+	__u32 key_num = cols_data->key_num;
+	__u64 row_count_size = 0;
+	__u32 i;
+
+	cols_data->index_num = setup->join_table.hash_table_index_num;
+
+	if (cols_data->index_num > HASH_TABLE_MAX_INDEX_NUM) {
+		WD_ERR("invalid: hash table index num is not supported!\n");
+		return -WD_EINVAL;
+	} else if (!cols_data->index_num) {
+		WD_INFO("Hash table index num is not set, set to default: 1!\n");
+		cols_data->index_num = HASH_TABLE_INDEX_NUM;
+	}
+
+	/* With a restriction on the col number, the sum lengths will not overflow. */
+	for (i = 0; i < key_num; i++)
+		row_count_size += get_data_type_size(key_data[i].hw_type, 0);
+
+	row_count_size = ALIGN(row_count_size, DAE_KEY_ALIGN_SIZE);
+	row_count_size += HASH_TABLE_HEAD_TAIL_SIZE +
+			  cols_data->index_num * HASH_TABLE_INDEX_SIZE;
+	if (row_count_size > DAE_MAX_ROW_SIZE) {
+		WD_ERR("invalid: hash table row size %llu, hash_table_index_num %u!\n",
+		       row_count_size, cols_data->index_num);
+		return -WD_EINVAL;
+	}
+
+	if (row_count_size <= ROW_SIZE32)
+		return ROW_SIZE32;
+
+	if (row_count_size <= ROW_SIZE64)
+		return ROW_SIZE64;
+
+	if (row_count_size <= ROW_SIZE128)
+		return ROW_SIZE128;
+
+	if (row_count_size <= ROW_SIZE256)
+		return ROW_SIZE256;
+
+	return ROW_SIZE512;
+}
+
+static void gather_get_batch_rowsize(struct join_gather_col_data *cols_data,
+				     struct wd_join_gather_sess_setup *setup,
+				     __u32 *batch_row_size)
+{
+	struct wd_gather_table_info *tables = setup->gather_tables;
+	struct hw_join_gather_data *gather_data;
+	__u32 row_count_size = 0;
+	__u32 n, i;
+
+	cols_data->gather_table_num = setup->gather_table_num;
+	for (n = 0; n < setup->gather_table_num; n++) {
+		row_count_size = 0;
+		gather_data = cols_data->gather_data[n];
+
+		/* With a restriction on the col number, the sum length will not overflow. */
+		for (i = 0; i < tables[n].cols_num; i++)
+			row_count_size += get_data_type_size(gather_data[i].hw_type,
+							     gather_data[i].data_info);
+
+		batch_row_size[n] = row_count_size + GATHER_ROW_BATCH_EMPTY_SIZE;
+	}
+}
+
+static int join_gather_fill_ctx(struct join_gather_ctx *ctx,
+				struct wd_join_gather_sess_setup *setup)
+{
+	struct join_gather_col_data *cols_data = &ctx->cols_data;
+	int ret;
+
+	if (setup->alg != WD_GATHER) {
+		ret = transfer_key_to_hw_type(cols_data, setup);
+		if (ret)
+			return ret;
+
+		ret = join_get_table_rowsize(cols_data, setup);
+		if (ret < 0)
+			return -WD_EINVAL;
+		ctx->hash_table_row_size = ret;
+	}
+
+	if (setup->alg != WD_JOIN) {
+		ret = transfer_data_to_hw_type(cols_data, setup);
+		if (ret)
+			return ret;
+
+		gather_get_batch_rowsize(cols_data, setup, ctx->batch_row_size);
+	}
+
+	return WD_SUCCESS;
+}
+
+static void join_gather_sess_priv_uninit(struct wd_alg_driver *drv, void *priv)
+{
+	struct join_gather_ctx *ctx = priv;
+
+	if (!ctx) {
+		WD_ERR("invalid: dae sess uninit priv is NULL!\n");
+		return;
+	}
+
+	pthread_spin_destroy(&ctx->lock);
+	free(ctx);
+}
+
+static int join_gather_sess_priv_init(struct wd_alg_driver *drv,
+				      struct wd_join_gather_sess_setup *setup, void **priv)
+{
+	struct wd_ctx_config_internal *config;
+	struct hisi_dae_ctx *dae_priv;
+	struct join_gather_ctx *ctx;
+	struct hisi_qp *qp;
+	handle_t h_qp;
+	int ret;
+
+	if (!drv || !drv->priv) {
+		WD_ERR("invalid: dae drv is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!setup || !priv) {
+		WD_ERR("invalid: dae sess priv is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	dae_priv = (struct hisi_dae_ctx *)drv->priv;
+	config = &dae_priv->config;
+	h_qp = (handle_t)wd_ctx_get_priv(config->ctxs[0].ctx);
+	qp = (struct hisi_qp *)h_qp;
+
+	ret = join_gather_param_check(setup, qp->q_info.hw_type);
+	if (ret)
+		return -WD_EINVAL;
+
+	ctx = calloc(1, sizeof(struct join_gather_ctx));
+	if (!ctx)
+		return -WD_ENOMEM;
+
+	ret = join_gather_fill_ctx(ctx, setup);
+	if (ret)
+		goto free_ctx;
+
+	ret = pthread_spin_init(&ctx->lock, PTHREAD_PROCESS_SHARED);
+	if (ret)
+		goto free_ctx;
+
+	*priv = ctx;
+
+	return WD_SUCCESS;
+
+free_ctx:
+	free(ctx);
+	return ret;
+}
+
+static void dae_uninit_qp_priv(handle_t h_qp)
+{
+	struct hisi_qp *qp = (struct hisi_qp *)h_qp;
+	struct dae_extend_addr *ext_addr = (struct dae_extend_addr *)qp->priv;
+
+	free(ext_addr->addr_list);
+	free(ext_addr->addr_status);
+	free(ext_addr->ext_sqe);
+	free(ext_addr);
+	qp->priv = NULL;
+}
+
+static int dae_init_qp_priv(handle_t h_qp)
+{
+	struct hisi_qp *qp = (struct hisi_qp *)h_qp;
+	__u16 sq_depth = qp->q_info.sq_depth;
+	struct dae_extend_addr *ext_addr;
+	int ret = -WD_ENOMEM;
+
+	ext_addr = calloc(1, sizeof(struct dae_extend_addr));
+	if (!ext_addr)
+		return ret;
+
+	ext_addr->ext_sqe = aligned_alloc(DAE_ADDR_ALIGN_SIZE, DAE_EXT_SQE_SIZE * sq_depth);
+	if (!ext_addr->ext_sqe)
+		goto free_ext_addr;
+
+	ext_addr->addr_status = calloc(1, sizeof(__u8) * sq_depth);
+	if (!ext_addr->addr_status)
+		goto free_ext_sqe;
+
+	ext_addr->addr_list = aligned_alloc(DAE_ADDR_ALIGN_SIZE,
+					    sizeof(struct dae_addr_list) * sq_depth);
+	if (!ext_addr->addr_list)
+		goto free_addr_status;
+
+	ext_addr->addr_num = sq_depth;
+	qp->priv = ext_addr;
+
+	return WD_SUCCESS;
+
+free_addr_status:
+	free(ext_addr->addr_status);
+free_ext_sqe:
+	free(ext_addr->ext_sqe);
+free_ext_addr:
+	free(ext_addr);
+
+	return ret;
+}
+
+static int join_get_table_row_size(struct wd_alg_driver *drv, void *param)
+{
+	struct join_gather_ctx *ctx = param;
+
+	if (!ctx)
+		return -WD_EINVAL;
+
+	return ctx->hash_table_row_size;
+}
+
+static int gather_get_batch_row_size(struct wd_alg_driver *drv, void *param,
+				     __u32 *row_size, __u32 size)
+{
+	struct join_gather_ctx *ctx = param;
+
+	if (!ctx)
+		return -WD_EINVAL;
+
+	if (!size || size >  DAE_MAX_TABLE_NUM * sizeof(__u32))
+		return -WD_EINVAL;
+
+	memcpy(row_size, ctx->batch_row_size, size);
+
+	return 0;
+}
+
+static __u32 dae_ext_table_rownum(void **ext_table, struct wd_dae_hash_table *table,
+				  __u32 row_size)
+{
+	__u64 tlb_size, tmp_size, row_num;
+	void *tmp_table;
+
+	/*
+	 * The first row of the extended hash table stores the hash table information,
+	 * and the second row stores the aggregated data. The 128-bytes aligned address
+	 * in the second row provides the optimal performance.
+	 */
+	tmp_table = PTR_ALIGN(table->ext_table, DAE_TABLE_ALIGN_SIZE);
+	tlb_size = (__u64)table->table_row_size * table->ext_table_row_num;
+	tmp_size = (__u64)(uintptr_t)tmp_table - (__u64)(uintptr_t)table->ext_table;
+	if (tmp_size >= tlb_size)
+		return 0;
+
+	row_num = (tlb_size - tmp_size) / row_size;
+	if (row_size == ROW_SIZE32) {
+		if (tmp_size >= row_size) {
+			tmp_table = (__u8 *)tmp_table - row_size;
+			row_num += 1;
+		} else {
+			/*
+			 * When row size is 32 bytes, the first 96 bytes are not used.
+			 * Ensure that the address of the second row is 128 bytes aligned.
+			 */
+			if (row_num <= HASH_TABLE_OFFSET_3ROW)
+				return 0;
+
+			tmp_table = (__u8 *)tmp_table + HASH_TABLE_OFFSET_3ROW * row_size;
+			row_num -= HASH_TABLE_OFFSET_3ROW;
+		}
+	} else if (row_size == ROW_SIZE64) {
+		if (tmp_size >= row_size) {
+			tmp_table = (__u8 *)tmp_table - row_size;
+			row_num += 1;
+		} else {
+			/*
+			 * When row size is 64 bytes, the first 64 bytes are not used.
+			 * Ensure that the address of the second row is 128 bytes aligned.
+			 */
+			if (row_num <= HASH_TABLE_OFFSET_1ROW)
+				return 0;
+
+			tmp_table = (__u8 *)tmp_table + HASH_TABLE_OFFSET_1ROW * row_size;
+			row_num -= HASH_TABLE_OFFSET_1ROW;
+		}
+	}
+
+	*ext_table = tmp_table;
+
+	return row_num;
+}
+
+static int dae_ext_table_init(struct join_gather_ctx *ctx,
+			      struct wd_dae_hash_table *table, bool is_rehash)
+{
+	struct hash_table_data *hw_table = &ctx->table_data;
+	__u64 ext_size = hw_table->ext_table_size;
+	__u32 row_size = ctx->hash_table_row_size;
+	__u64 tlb_size, row_num;
+	void *ext_table;
+	__u8 *ext_valid;
+	__u64 *ext_row;
+
+	row_num = dae_ext_table_rownum(&ext_table, table, row_size);
+	if (row_num <= 1) {
+		WD_ERR("invalid: extend table row num is less than device need!\n");
+		return -WD_EINVAL;
+	}
+
+	tlb_size = row_num * row_size;
+	if (is_rehash && tlb_size <= ext_size) {
+		WD_ERR("rehash extend table size %llu is not longer than current %llu!\n",
+			tlb_size, ext_size);
+		return -WD_EINVAL;
+	}
+
+	/*
+	 * If table has been initialized, save the previous data
+	 * before replacing the new table.
+	 */
+	if (is_rehash)
+		memcpy(&ctx->rehash_table, hw_table, sizeof(struct hash_table_data));
+
+	/* Initialize the extend table value. */
+	memset(ext_table, 0, tlb_size);
+	ext_valid = (__u8 *)ext_table + HASH_EXT_TABLE_INVALID_OFFSET;
+	*ext_valid = HASH_EXT_TABLE_VALID;
+	ext_row = (__u64 *)ext_table + 1;
+	*ext_row = row_num - 1;
+
+	hw_table->ext_table = ext_table;
+	hw_table->ext_table_size = tlb_size;
+
+	return WD_SUCCESS;
+}
+
+static int dae_std_table_init(struct hash_table_data *hw_table,
+			      struct wd_dae_hash_table *table, __u32 row_size)
+{
+	__u64 tlb_size, row_num, tmp_size;
+
+	/*
+	 * Hash table address must be 128-bytes aligned, and the number
+	 * of rows in a standard hash table must be a power of 2.
+	 */
+	hw_table->std_table = PTR_ALIGN(table->std_table, DAE_TABLE_ALIGN_SIZE);
+	tlb_size = (__u64)table->table_row_size * table->std_table_row_num;
+	tmp_size = (__u64)(uintptr_t)hw_table->std_table - (__u64)(uintptr_t)table->std_table;
+	if (tmp_size >= tlb_size) {
+		WD_ERR("invalid: after aligned, standard table size is less than 0!\n");
+		return -WD_EINVAL;
+	}
+
+	row_num = (tlb_size - tmp_size) / row_size;
+	if (!row_num) {
+		WD_ERR("invalid: standard table row num is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	hw_table->table_width = (__u32)log2(row_num);
+	if (hw_table->table_width < HASH_TABLE_MIN_WIDTH ||
+	    hw_table->table_width > HASH_TABLE_MAX_WIDTH) {
+		WD_ERR("standard table width %u is out of device support range %d~%d!\n",
+			hw_table->table_width, HASH_TABLE_MIN_WIDTH, HASH_TABLE_MAX_WIDTH);
+		return -WD_EINVAL;
+	}
+
+	row_num = (__u64)pow(HASH_TABLE_WITDH_POWER, hw_table->table_width);
+	hw_table->std_table_size = row_num * row_size;
+	memset(hw_table->std_table, 0, hw_table->std_table_size);
+
+	return WD_SUCCESS;
+}
+
+static int join_hash_table_init(struct wd_alg_driver *drv,
+				struct wd_dae_hash_table *table, void *priv)
+{
+	struct join_gather_ctx *ctx = priv;
+	struct hash_table_data *hw_table;
+	bool is_rehash = false;
+	int ret;
+
+	if (!ctx || !table)
+		return -WD_EINVAL;
+
+	if (!ctx->hash_table_row_size || ctx->hash_table_row_size > table->table_row_size) {
+		WD_ERR("invalid: row size %u is error, device need %u!\n",
+			table->table_row_size, ctx->hash_table_row_size);
+		return -WD_EINVAL;
+	}
+
+	/* hash_std_table is checked by caller */
+	if (!table->ext_table || !table->ext_table_row_num) {
+		WD_ERR("invalid: hash extend table is null!\n");
+		return -WD_EINVAL;
+	}
+
+	hw_table = &ctx->table_data;
+	if (hw_table->std_table_size)
+		is_rehash = true;
+
+	ret = dae_ext_table_init(ctx, table, is_rehash);
+	if (ret)
+		return ret;
+
+	ret = dae_std_table_init(hw_table, table, ctx->hash_table_row_size);
+	if (ret)
+		goto update_table;
+
+	return WD_SUCCESS;
+
+update_table:
+	if (is_rehash)
+		memcpy(hw_table, &ctx->rehash_table, sizeof(struct hash_table_data));
+	else
+		memset(hw_table, 0, sizeof(struct hash_table_data));
+	return ret;
+}
+
+static int dae_init(struct wd_alg_driver *drv, void *conf)
+{
+	struct wd_ctx_config_internal *config = conf;
+	struct hisi_qm_priv qm_priv;
+	struct hisi_dae_ctx *priv;
+	handle_t h_qp = 0;
+	handle_t h_ctx;
+	__u32 i, j;
+	int ret;
+
+	if (!config || !config->ctx_num) {
+		WD_ERR("invalid: dae init config is null or ctx num is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	priv = malloc(sizeof(struct hisi_dae_ctx));
+	if (!priv)
+		return -WD_ENOMEM;
+
+	qm_priv.op_type = DAE_SQC_ALG_TYPE;
+	qm_priv.sqe_size = sizeof(struct dae_sqe);
+	/* Allocate qp for each context */
+	for (i = 0; i < config->ctx_num; i++) {
+		h_ctx = config->ctxs[i].ctx;
+		qm_priv.qp_mode = config->ctxs[i].ctx_mode;
+		/* Setting the epoll en to 0 for ASYNC ctx */
+		qm_priv.epoll_en = (qm_priv.qp_mode == CTX_MODE_SYNC) ?
+				   config->epoll_en : 0;
+		qm_priv.idx = i;
+		h_qp = hisi_qm_alloc_qp(&qm_priv, h_ctx);
+		if (!h_qp) {
+			ret = -WD_ENOMEM;
+			goto out;
+		}
+		config->ctxs[i].sqn = qm_priv.sqn;
+		ret = dae_init_qp_priv(h_qp);
+		if (ret)
+			goto free_h_qp;
+	}
+	memcpy(&priv->config, config, sizeof(struct wd_ctx_config_internal));
+	drv->priv = priv;
+
+	return WD_SUCCESS;
+
+free_h_qp:
+	hisi_qm_free_qp(h_qp);
+out:
+	for (j = 0; j < i; j++) {
+		h_qp = (handle_t)wd_ctx_get_priv(config->ctxs[j].ctx);
+		dae_uninit_qp_priv(h_qp);
+		hisi_qm_free_qp(h_qp);
+	}
+	free(priv);
+	return ret;
+}
+
+static void dae_exit(struct wd_alg_driver *drv)
+{
+	struct wd_ctx_config_internal *config;
+	struct hisi_dae_ctx *priv;
+	handle_t h_qp;
+	__u32 i;
+
+	if (!drv || !drv->priv)
+		return;
+
+	priv = (struct hisi_dae_ctx *)drv->priv;
+	config = &priv->config;
+	for (i = 0; i < config->ctx_num; i++) {
+		h_qp = (handle_t)wd_ctx_get_priv(config->ctxs[i].ctx);
+		dae_uninit_qp_priv(h_qp);
+		hisi_qm_free_qp(h_qp);
+	}
+
+	free(priv);
+	drv->priv = NULL;
+}
+
+static int join_gather_get_extend_ops(void *ops)
+{
+	struct wd_join_gather_ops *join_gather_ops = (struct wd_join_gather_ops *)ops;
+
+	if (!join_gather_ops)
+		return -WD_EINVAL;
+
+	join_gather_ops->get_table_row_size = join_get_table_row_size;
+	join_gather_ops->get_batch_row_size = gather_get_batch_row_size;
+	join_gather_ops->hash_table_init = join_hash_table_init;
+	join_gather_ops->sess_init = join_gather_sess_priv_init;
+	join_gather_ops->sess_uninit = join_gather_sess_priv_uninit;
+
+	return WD_SUCCESS;
+}
+
+
+#define GEN_JOIN_GATHER_DRIVER(dae_alg_name) \
+{\
+	.drv_name = "hisi_zip",\
+	.alg_name = (dae_alg_name),\
+	.calc_type = UADK_ALG_HW,\
+	.priority = 100,\
+	.queue_num = DAE_CTX_Q_NUM_DEF,\
+	.op_type_num = 1,\
+	.fallback = 0,\
+	.init = dae_init,\
+	.exit = dae_exit,\
+	.send = join_gather_send,\
+	.recv = join_gather_recv,\
+	.get_extend_ops = join_gather_get_extend_ops,\
+}
+
+static struct wd_alg_driver join_gather_driver[] = {
+	GEN_JOIN_GATHER_DRIVER("hashjoin"),
+	GEN_JOIN_GATHER_DRIVER("gather"),
+	GEN_JOIN_GATHER_DRIVER("join-gather"),
+};
+
+#ifdef WD_STATIC_DRV
+void hisi_dae_join_gather_probe(void)
+#else
+static void __attribute__((constructor)) hisi_dae_join_gather_probe(void)
+#endif
+{
+	__u32 alg_num = ARRAY_SIZE(join_gather_driver);
+	int ret;
+	__u32 i;
+
+	WD_INFO("Info: register DAE hashjoin and gather alg drivers!\n");
+	for (i = 0; i < alg_num; i++) {
+		ret = wd_alg_driver_register(&join_gather_driver[i]);
+		if (ret && ret != -WD_ENODEV)
+			WD_ERR("Error: register %s failed!\n",
+			       join_gather_driver[i].alg_name);
+	}
+}
+
+#ifdef WD_STATIC_DRV
+void hisi_dae_join_gather_remove(void)
+#else
+static void __attribute__((destructor)) hisi_dae_join_gather_remove(void)
+#endif
+{
+	__u32 alg_num = ARRAY_SIZE(join_gather_driver);
+	__u32 i;
+
+	WD_INFO("Info: unregister DAE alg drivers!\n");
+	for (i = 0; i < alg_num; i++)
+		wd_alg_driver_unregister(&join_gather_driver[i]);
+}
diff --git a/include/drv/wd_join_gather_drv.h b/include/drv/wd_join_gather_drv.h
new file mode 100644
index 00000000..80fb9322
--- /dev/null
+++ b/include/drv/wd_join_gather_drv.h
@@ -0,0 +1,52 @@
+/* SPDX-License-Identifier: Apache-2.0 */
+/*
+ * Copyright 2025 Huawei Technologies Co.,Ltd. All rights reserved.
+ */
+
+#ifndef __WD_JOIN_GATHER_DRV_H
+#define __WD_JOIN_GATHER_DRV_H
+
+#include <asm/types.h>
+#include "wd_join_gather.h"
+#include "wd_util.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct wd_join_gather_msg {
+	__u32 tag;
+	__u32 key_cols_num;
+	__u32 result;
+	__u32 input_row_num;
+	__u32 output_row_num;
+	__u32 consumed_row_num;
+	__u32 produced_row_num;
+	enum wd_join_gather_op_type op_type;
+	enum multi_batch_index_type index_type;
+	bool output_done;
+	bool key_out_en;
+	bool multi_batch_en;
+	struct wd_join_gather_req req;
+	struct wd_dae_hash_table hash_table;
+	void *priv;
+};
+
+struct wd_join_gather_ops {
+	int (*get_table_row_size)(struct wd_alg_driver *drv, void *priv);
+	int (*get_batch_row_size)(struct wd_alg_driver *drv, void *priv,
+				  __u32 *batch_row_size, __u32 size);
+	int (*sess_init)(struct wd_alg_driver *drv,
+			 struct wd_join_gather_sess_setup *setup, void **priv);
+	void (*sess_uninit)(struct wd_alg_driver *drv, void *priv);
+	int (*hash_table_init)(struct wd_alg_driver *drv,
+			       struct wd_dae_hash_table *hash_table, void *priv);
+};
+
+struct wd_join_gather_msg *wd_join_gather_get_msg(__u32 idx, __u32 tag);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __WD_JOIN_GATHER_DRV_H */
diff --git a/include/wd_alg.h b/include/wd_alg.h
index 441b3bef..2fc350af 100644
--- a/include/wd_alg.h
+++ b/include/wd_alg.h
@@ -205,12 +205,14 @@ void hisi_hpre_probe(void);
 void hisi_zip_probe(void);
 void hisi_dae_probe(void);
 void hisi_udma_probe(void);
+void hisi_dae_join_gather_probe(void);
 
 void hisi_sec2_remove(void);
 void hisi_hpre_remove(void);
 void hisi_zip_remove(void);
 void hisi_dae_remove(void);
 void hisi_udma_remove(void);
+void hisi_dae_join_gather_remove(void);
 
 #endif
 
diff --git a/include/wd_dae.h b/include/wd_dae.h
index aa9f966c..64f17dc4 100644
--- a/include/wd_dae.h
+++ b/include/wd_dae.h
@@ -57,6 +57,18 @@ struct wd_dae_col_addr {
 	__u64 offset_size;
 };
 
+/**
+ * wd_dae_row_addr - information of row memory.
+ * @addr: The start address of row memory.
+ * @row_size: Memory size occupied by a row.
+ * @row_num: Total number of rows.
+ */
+struct wd_dae_row_addr {
+	void *addr;
+	__u32 row_size;
+	__u32 row_num;
+};
+
 /**
  * wd_dae_hash_table - Hash table information of DAE.
  * @std_table: Address of standard hash table.
diff --git a/include/wd_join_gather.h b/include/wd_join_gather.h
new file mode 100644
index 00000000..4962ee35
--- /dev/null
+++ b/include/wd_join_gather.h
@@ -0,0 +1,352 @@
+/* SPDX-License-Identifier: Apache-2.0 */
+/*
+ * Copyright 2025 Huawei Technologies Co.,Ltd. All rights reserved.
+ */
+
+#ifndef __WD_JOIN_GATHER_H
+#define __WD_JOIN_GATHER_H
+
+#include <dlfcn.h>
+#include <asm/types.h>
+#include "wd_dae.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+enum wd_join_gather_alg {
+	WD_JOIN,
+	WD_GATHER,
+	WD_JOIN_GATHER,
+	WD_JOIN_GATHER_ALG_MAX,
+};
+
+/**
+ * wd_join_gather_op_type - operation type for hash join and gather.
+ */
+enum wd_join_gather_op_type {
+	WD_JOIN_BUILD_HASH,
+	WD_JOIN_PROBE,
+	WD_JOIN_REHASH,
+	WD_GATHER_CONVERT,
+	WD_GATHER_COMPLETE,
+	WD_JOIN_GATHER_OP_TYPE_MAX,
+};
+
+/**
+ * wd_join_gather_task_error_type - hash join and gather task error type.
+ */
+enum wd_join_gather_task_error_type {
+	WD_JOIN_GATHER_TASK_DONE,
+	WD_JOIN_GATHER_IN_EPARA,
+	WD_JOIN_GATHER_NEED_REHASH,
+	WD_JOIN_GATHER_INVALID_HASH_TABLE,
+	WD_JOIN_GATHER_PARSE_ERROR,
+	WD_JOIN_GATHER_BUS_ERROR,
+};
+
+enum multi_batch_index_type {
+	WD_BATCH_NUMBER_INDEX,
+	WD_BATCH_ADDR_INDEX,
+	WD_BATCH_INDEX_TYPE_MAX,
+};
+
+/**
+ * wd_join_gather_col_info - column information.
+ * @data_type: column data type.
+ * @data_info: For CHAR, it is size of data, at least 1B.
+ * For DECIMAL, it is precision of data, high 8 bit: decimal part precision,
+ * low 8 bit: the whole data precision.
+ * @has_empty: indicates whether the column contains empty data.
+ */
+struct wd_join_gather_col_info {
+	enum wd_dae_data_type data_type;
+	__u16	data_info;
+	bool	has_empty;
+};
+
+/**
+ * wd_gather_table_info - gather table information.
+ * @cols: Information of gather table columns.
+ * @cols_num: Number of gather table columns.
+ * @is_multi_batch: indicates single or multi batch task.
+ */
+struct wd_gather_table_info {
+	struct wd_join_gather_col_info *cols;
+	__u32	cols_num;
+	bool	is_multi_batch;
+};
+
+/**
+ * wd_join_table_info - join table information.
+ * @build_key_cols: Information of build table key columns.
+ * @probe_key_cols: Information of probe table key columns.
+ * @build_key_cols_num: Number of build table key columns.
+ * @probe_key_cols_num: Number of probe table key columns.
+ * @key_output_enable: Indicates whether output key columns.
+ * @hash_table_index_num: Number of original rows can be stored
+ * in each row of a hash table.
+ */
+struct wd_join_table_info {
+	struct wd_join_gather_col_info *build_key_cols;
+	struct wd_join_gather_col_info *probe_key_cols;
+	__u32	build_key_cols_num;
+	__u32	probe_key_cols_num;
+	bool	key_output_enable;
+	__u32	hash_table_index_num;
+};
+
+/**
+ * wd_join_gather_sess_setup - Hash join and gather session setup information.
+ * @join_table: Information of join table.
+ * @gather_tables: Information of gather table.
+ * @gather_table_num: Number of gather table.
+ * @alg: Alg for this session.
+ * @index_type: Indicates the index type, 0 for batch number and row number,
+ * 1 for batch address and row number.
+ * @charset_info: Charset information
+ * @sched_param: Parameters of the scheduling policy,
+ * usually allocated according to struct sched_params.
+ */
+struct wd_join_gather_sess_setup {
+	struct wd_join_table_info	join_table;
+	struct wd_gather_table_info	*gather_tables;
+	__u32 gather_table_num;
+
+	enum wd_join_gather_alg		alg;
+	enum multi_batch_index_type	index_type;
+	struct wd_dae_charset		charset_info;
+	void	*sched_param;
+};
+
+struct wd_join_gather_req;
+typedef void *wd_join_gather_cb_t(struct wd_join_gather_req *req, void *cb_param);
+
+/**
+ * wd_probe_out_info - Hash join probe output info.
+ * @build_index: address information of multi batch index.
+ * @probe_index: address information of single batch index.
+ * @breakpoint: address information of probe breakpoint.
+ * @key_cols: address information of output key columns.
+ * @key_cols_num: number of output key columns.
+ */
+struct wd_probe_out_info {
+	struct wd_dae_row_addr build_index;
+	struct wd_dae_row_addr probe_index;
+	struct wd_dae_row_addr breakpoint;
+	struct wd_dae_col_addr *key_cols;
+	__u32 key_cols_num;
+};
+
+/**
+ * wd_join_req - Hash join request.
+ * @build_batch_addr: Row-storaged batch address, the batch is used to store build
+ * table data cols in row format. This field is only used for batch addr index.
+ *
+ * @probe_output: The information for hash join probe stage.
+ * @key_cols: key columns from build table or probe table.
+ * @key_cols_num: key columns number.
+ * @batch_row_offset: Indicates the start row number of the input column.
+ * @build_batch_index: build table batch index, start from 0.
+ */
+struct wd_join_req {
+	struct wd_dae_row_addr		build_batch_addr;
+	struct wd_probe_out_info	probe_output;
+	struct wd_dae_col_addr		*key_cols;
+	__u32 key_cols_num;
+	__u32 batch_row_offset;
+	__u32 build_batch_index;
+};
+
+/**
+ * wd_row_batch_info - Information of some row-storaged batchs.
+ * @batch_addr: Addr list of row batchs.
+ * @batch_row_size: Row size of each row batch.
+ * @batch_row_num: Row number of each row batch.
+ * @batch_num: Total number of row batchs.
+ */
+struct wd_row_batch_info {
+	void **batch_addr;
+	__u32 *batch_row_size;
+	__u32 *batch_row_num;
+	__u32 batch_num;
+};
+
+/**
+ * wd_gather_req - Hash join and gather operation request.
+ * @index: address information of multi batch index or single batch index.
+ * @row_batchs: address information of row batchs.
+ * @data_cols: data columns from gather table.
+ * @data_cols_num: columns number from gather table.
+ * @table_index: The table index from the session's gather_tables to do tasks.
+ */
+struct wd_gather_req {
+	struct wd_dae_row_addr index;
+	struct wd_row_batch_info row_batchs;
+	struct wd_dae_col_addr *data_cols;
+	__u32 data_cols_num;
+	__u32 table_index;
+};
+
+/**
+ * wd_join_gather_req - Hash join and gather operation request.
+ * @op_type: The operation type for hash join or gather task.
+ * @join_req: The request for hash join.
+ * @gather_req: The request for gather.
+ * @input_row_num: Row count of input column.
+ * @output_row_num: Expected row count of output column.
+ * @consumed_row_num: Row count of input data that has been processed.
+ * @produced_row_num: Real row count of output column.
+ * @cb: Callback function for the asynchronous mode.
+ * @cb_param: Parameters of the callback function.
+ * @state: Error information written back by the hardware.
+ * @output_done: For rehash, it indicates whether all data in hash table has been output.
+ * For probe task, it indicates whether all data of one probe batch has been processed.
+ * @priv: Private data from user(reserved).
+ */
+struct wd_join_gather_req {
+	/* user fill-in fields */
+	enum wd_join_gather_op_type	op_type;
+	struct wd_join_req		join_req;
+	struct wd_gather_req		gather_req;
+	__u32 input_row_num;
+	__u32 output_row_num;
+	wd_join_gather_cb_t *cb;
+	void *cb_param;
+	void *priv;
+
+	/* uadk driver writeback fields */
+	enum wd_join_gather_task_error_type state;
+	__u32	consumed_row_num;
+	__u32	produced_row_num;
+	bool	output_done;
+};
+
+/**
+ * wd_join_gather_init() - A simplify interface to initializate uadk.
+ * Users just need to descripe the deployment of business scenarios.
+ * Then the initialization will request appropriate
+ * resources to support the business scenarios.
+ * To make the initializate simpler, ctx_params support set NULL.
+ * And then the function will set them as driver's default.
+ *
+ * @alg: Supported algorithms: hashjoin, gather, join-gather.
+ * @sched_type: The scheduling type users want to use.
+ * @task_type: Task types, including soft computing, hardware and hybrid computing.
+ * @ctx_params: The ctxs resources users want to use. Include per operation
+ * type ctx numbers and business process run numa.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_gather_init(char *alg, __u32 sched_type, int task_type,
+			struct wd_ctx_params *ctx_params);
+
+/**
+ * wd_join_gather_uninit() - Uninitialise ctx configuration and scheduler.
+ */
+void wd_join_gather_uninit(void);
+
+/**
+ * wd_join_gather_alloc_sess() - Allocate a hash join or gather session
+ * @setup: Parameters to setup this session.
+ *
+ * Return 0 if fail and others if succeed.
+ */
+handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup);
+
+/**
+ * wd_join_gather_free_sess() - Free a hash join or gather session
+ * @sess: The session need to be freed.
+ */
+void wd_join_gather_free_sess(handle_t h_sess);
+
+/**
+ * wd_join_set_hash_table() - Set hash table to the wd session
+ * @sess, Session to be initialized.
+ * @info, Hash table information to set.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info);
+
+/**
+ * wd_join_build_hash_sync()/wd_join_build_hash_async() - Build the hash table.
+ * @sess: Wd session
+ * @req: Operational data.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_build_hash_sync(handle_t h_sess, struct wd_join_gather_req *req);
+int wd_join_build_hash_async(handle_t h_sess, struct wd_join_gather_req *req);
+
+/**
+ * wd_join_probe_sync()/wd_join_probe_async() - Probe and output the index or key.
+ * @sess: Wd session
+ * @req: Operational data.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_probe_sync(handle_t h_sess, struct wd_join_gather_req *req);
+int wd_join_probe_async(handle_t h_sess, struct wd_join_gather_req *req);
+
+/**
+ * wd_gather_convert_sync()/wd_gather_convert_async() - Convert a column batch to a row batch.
+ * @sess: Wd session
+ * @req: Operational data.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_gather_convert_sync(handle_t h_sess, struct wd_join_gather_req *req);
+int wd_gather_convert_async(handle_t h_sess, struct wd_join_gather_req *req);
+
+/**
+ * wd_gather_complete_sync()/wd_gather_complete_async() - map the index with a row batch
+ * and output the result to a column batch.
+ * @sess: Wd session
+ * @req: Operational data.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_gather_complete_sync(handle_t h_sess, struct wd_join_gather_req *req);
+int wd_gather_complete_async(handle_t h_sess, struct wd_join_gather_req *req);
+
+/**
+ * wd_join_rehash_sync - Rehash operation, only the synchronous mode is supported.
+ * @sess: Wd hash join session
+ * @req: Operational data.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_rehash_sync(handle_t h_sess, struct wd_join_gather_req *req);
+
+/**
+ * wd_join_gather_poll() - Poll finished request.
+ * This function will call poll_policy function which is registered to wd
+ * by user.
+ *
+ * Return 0 if succeed and others if fail.
+ */
+int wd_join_gather_poll(__u32 expt, __u32 *count);
+
+/**
+ * wd_join_get_table_rowsize - Get the hash table's row size.
+ * @h_sess: Wd session handler.
+ *
+ * Return negative value if fail and others if succeed.
+ */
+int wd_join_get_table_rowsize(handle_t h_sess);
+
+/**
+ * wd_gather_get_batch_rowsize - Get the batch row size.
+ * @h_sess: Wd session handler.
+ * @table_index: The table index from the session's gather_tables.
+ *
+ * Return negative value if fail and others if succeed.
+ */
+int wd_gather_get_batch_rowsize(handle_t h_sess, __u8 table_index);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __WD_JOIN_GATHER_H */
diff --git a/include/wd_util.h b/include/wd_util.h
index bbb18a7c..4a5204de 100644
--- a/include/wd_util.h
+++ b/include/wd_util.h
@@ -43,6 +43,7 @@ enum wd_type {
 	WD_ECC_TYPE,
 	WD_AGG_TYPE,
 	WD_UDMA_TYPE,
+	WD_JOIN_GATHER_TYPE,
 	WD_TYPE_MAX,
 };
 
diff --git a/libwd_dae.map b/libwd_dae.map
index 6597ff98..f3b06337 100644
--- a/libwd_dae.map
+++ b/libwd_dae.map
@@ -1,5 +1,24 @@
 UADK_DAE_2.0 {
 global:
+	wd_join_gather_alloc_sess;
+	wd_join_gather_free_sess;
+	wd_join_get_table_rowsize;
+	wd_gather_get_batch_rowsize;
+	wd_join_set_hash_table;
+	wd_join_gather_init;
+	wd_join_gather_uninit;
+	wd_join_build_hash_sync;
+	wd_join_build_hash_async;
+	wd_join_probe_sync;
+	wd_join_probe_async;
+	wd_join_rehash_sync;
+	wd_join_gather_get_msg;
+	wd_join_gather_poll;
+	wd_gather_convert_sync;
+	wd_gather_complete_sync;
+	wd_gather_convert_async;
+	wd_gather_complete_async;
+
 	wd_agg_alloc_sess;
 	wd_agg_free_sess;
 	wd_agg_get_table_rowsize;
diff --git a/wd_join_gather.c b/wd_join_gather.c
new file mode 100644
index 00000000..454e40e9
--- /dev/null
+++ b/wd_join_gather.c
@@ -0,0 +1,1821 @@
+/* SPDX-License-Identifier: Apache-2.0 */
+/*
+ * Copyright 2025 Huawei Technologies Co.,Ltd. All rights reserved.
+ */
+
+#include <stdlib.h>
+#include <pthread.h>
+#include <sched.h>
+#include <limits.h>
+#include "include/drv/wd_join_gather_drv.h"
+#include "wd_join_gather.h"
+
+#define DECIMAL_PRECISION_OFFSET	8
+#define DAE_INT_SIZE			4
+#define DAE_LONG_SIZE			8
+#define DAE_LONG_DECIMAL_SIZE		16
+
+/* Sum of the max row number of standard and external hash table */
+#define MAX_HASH_TABLE_ROW_NUM		0x1FFFFFFFE
+
+enum wd_join_sess_state {
+	WD_JOIN_SESS_UNINIT, /* Uninit session */
+	WD_JOIN_SESS_INIT, /* Hash table has been set */
+	WD_JOIN_SESS_BUILD_HASH, /* Input stage has started */
+	WD_JOIN_SESS_PREPARE_REHASH, /* New hash table has been set */
+	WD_JOIN_SESS_REHASH, /* Rehash stage has started */
+	WD_JOIN_SESS_PROBE, /* Output stage has started */
+};
+
+struct wd_join_gather_setting {
+	enum wd_status status;
+	struct wd_ctx_config_internal config;
+	struct wd_sched sched;
+	struct wd_async_msg_pool pool;
+	struct wd_alg_driver *driver;
+	void *priv;
+	void *dlhandle;
+	void *dlh_list;
+};
+
+struct wd_join_cols_conf {
+	struct wd_join_gather_col_info *cols;
+	__u64 *data_size;
+	__u32 cols_num;
+	bool key_output_enable;
+};
+
+struct wd_gather_tables_conf {
+	struct wd_gather_table_info *tables;
+	__u32 *batch_row_size;
+	__u64 **data_size;
+	__u32 table_num;
+};
+
+struct wd_join_gather_sess {
+	enum multi_batch_index_type index_type;
+	enum wd_join_sess_state state;
+	enum wd_join_gather_alg alg;
+	struct wd_join_gather_ops ops;
+	struct wd_join_cols_conf join_conf;
+	struct wd_gather_tables_conf gather_conf;
+	struct wd_dae_hash_table hash_table;
+	wd_dev_mask_t *dev_mask;
+	void *sched_key;
+	void *priv;
+};
+
+static const char *wd_join_gather_alg[WD_JOIN_GATHER_ALG_MAX] = {
+	"hashjoin", "gather", "join-gather"
+};
+
+static struct wd_init_attrs wd_join_gather_init_attrs;
+static struct wd_join_gather_setting wd_join_gather_setting;
+static int wd_join_gather_poll_ctx(__u32 idx, __u32 expt, __u32 *count);
+
+static void wd_join_gather_close_driver(void)
+{
+#ifndef WD_STATIC_DRV
+	wd_dlclose_drv(wd_join_gather_setting.dlh_list);
+#else
+	wd_release_drv(wd_join_gather_setting.driver);
+	hisi_dae_join_gather_remove();
+#endif
+}
+
+static int wd_join_gather_open_driver(void)
+{
+#ifndef WD_STATIC_DRV
+	/*
+	 * Driver lib file path could set by env param.
+	 * then open tham by wd_dlopen_drv()
+	 * use NULL means dynamic query path
+	 */
+	wd_join_gather_setting.dlh_list = wd_dlopen_drv(NULL);
+	if (!wd_join_gather_setting.dlh_list) {
+		WD_ERR("fail to open driver lib files.\n");
+		return -WD_EINVAL;
+	}
+#else
+	hisi_dae_join_gather_probe();
+#endif
+	return WD_SUCCESS;
+}
+
+static bool wd_join_gather_check_inner(void)
+{
+	struct uacce_dev_list *list;
+
+	list = wd_get_accel_list("hashjoin");
+	if (!list)
+		goto out;
+	wd_free_list_accels(list);
+
+	list = wd_get_accel_list("gather");
+	if (!list)
+		goto out;
+	wd_free_list_accels(list);
+
+	return true;
+out:
+	WD_ERR("invalid: the device cannot support hashjoin and gather!\n");
+	return false;
+}
+
+static bool wd_join_gather_alg_check(const char *alg_name)
+{
+	__u32 i;
+
+	/* Check for the virtual algorithms */
+	if (!strcmp(alg_name, "join-gather"))
+		return wd_join_gather_check_inner();
+
+	for (i = 0; i < WD_JOIN_GATHER_ALG_MAX; i++) {
+		/* Some algorithms do not support all modes */
+		if (!wd_join_gather_alg[i] || !strlen(wd_join_gather_alg[i]))
+			continue;
+		if (!strcmp(alg_name, wd_join_gather_alg[i]))
+			return true;
+	}
+
+	return false;
+}
+
+static int check_col_data_info(enum wd_dae_data_type type, __u16 col_data_info)
+{
+	__u8 all_precision, decimal_precision;
+
+	switch (type) {
+	case WD_DAE_DATE:
+	case WD_DAE_INT:
+	case WD_DAE_LONG:
+	case WD_DAE_VARCHAR:
+		break;
+	case WD_DAE_SHORT_DECIMAL:
+	case WD_DAE_LONG_DECIMAL:
+		/* High 8 bit: decimal part precision, low 8 bit: the whole data precision */
+		all_precision = col_data_info;
+		decimal_precision = col_data_info >> DECIMAL_PRECISION_OFFSET;
+		if (!all_precision || decimal_precision > all_precision) {
+			WD_ERR("failed to check data precision, all: %u, decimal: %u!\n",
+			       all_precision, decimal_precision);
+			return -WD_EINVAL;
+		}
+		break;
+	case WD_DAE_CHAR:
+		if (!col_data_info) {
+			WD_ERR("invalid: char length is zero!\n");
+			return -WD_EINVAL;
+		}
+		break;
+	default:
+		WD_ERR("invalid: data type %u is not supported!\n", type);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int get_data_type_size(enum wd_dae_data_type type, __u16 col_data_info,
+			      __u64 *col, __u32 idx)
+{
+	switch (type) {
+	case WD_DAE_DATE:
+	case WD_DAE_INT:
+		col[idx] = DAE_INT_SIZE;
+		break;
+	case WD_DAE_LONG:
+	case WD_DAE_SHORT_DECIMAL:
+		col[idx] = DAE_LONG_SIZE;
+		break;
+	case WD_DAE_LONG_DECIMAL:
+		col[idx] = DAE_LONG_DECIMAL_SIZE;
+		break;
+	case WD_DAE_CHAR:
+		col[idx] = col_data_info;
+		break;
+	case WD_DAE_VARCHAR:
+		col[idx] = 0;
+		break;
+	default:
+		return -WD_EINVAL;
+	}
+	return WD_SUCCESS;
+}
+
+static int check_key_cols_info(struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_join_table_info *table = &setup->join_table;
+	struct wd_join_gather_col_info *build = table->build_key_cols;
+	__u32 i;
+	int ret;
+
+	if (table->build_key_cols_num != table->probe_key_cols_num) {
+		WD_ERR("invalid: build key_cols_num: %u, probe key_cols_num: %u!\n",
+		       table->build_key_cols_num, table->probe_key_cols_num);
+		return -WD_EINVAL;
+	}
+
+	ret = memcmp(table->build_key_cols, table->probe_key_cols,
+		     table->build_key_cols_num * sizeof(struct wd_join_gather_col_info));
+	if (ret) {
+		WD_ERR("invalid: build and probe table key infomation is not same!\n");
+		return -WD_EINVAL;
+	}
+
+	for (i = 0; i < table->build_key_cols_num; i++) {
+		if (!build[i].has_empty) {
+			WD_ERR("invalid: key col has no empty data! col: %u\n", i);
+			return ret;
+		}
+		ret = check_col_data_info(build[i].data_type, build[i].data_info);
+		if (ret) {
+			WD_ERR("failed to check key col data info! col: %u\n", i);
+			return ret;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_check_params(struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_join_table_info *table = &setup->join_table;
+
+	if (!table->build_key_cols_num || !table->build_key_cols) {
+		WD_ERR("invalid: build key cols is NULL or key_cols_num is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!table->probe_key_cols_num || !table->probe_key_cols) {
+		WD_ERR("invalid: probe key cols is NULL or key_cols_num is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	if (setup->index_type >= WD_BATCH_INDEX_TYPE_MAX) {
+		WD_ERR("failed to check batch index type!\n");
+		return -WD_EINVAL;
+	}
+
+	if (check_key_cols_info(setup)) {
+		WD_ERR("failed to check join setup key cols info!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_gather_check_params(struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_gather_table_info *table = setup->gather_tables;
+	struct wd_join_gather_col_info *col;
+	__u32 i, j;
+	int ret;
+
+	if (!setup->gather_tables || !setup->gather_table_num) {
+		WD_ERR("invalid: gather table is NULL, table num: %u\n", setup->gather_table_num);
+		return -WD_EINVAL;
+	}
+
+	if (setup->index_type >= WD_BATCH_INDEX_TYPE_MAX) {
+		WD_ERR("failed to check gather batch index type!\n");
+		return -WD_EINVAL;
+	}
+
+	for (i = 0; i < setup->gather_table_num; i++) {
+		if (!table[i].cols || !table[i].cols_num) {
+			WD_ERR("failed to check gather table cols, num: %u\n", table[i].cols_num);
+			return -WD_EINVAL;
+		}
+		col = table[i].cols;
+		for (j = 0; j < table[i].cols_num; j++) {
+			ret = check_col_data_info(col[j].data_type, col[j].data_info);
+			if (ret) {
+				WD_ERR("failed to check gather info! col: %u, table: %u\n", j, i);
+				return ret;
+			}
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_gather_check_params(struct wd_join_gather_sess_setup *setup)
+{
+	int ret;
+
+	if (!setup) {
+		WD_ERR("invalid: hashjoin or gather sess setup is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	switch (setup->alg) {
+	case WD_JOIN:
+		return wd_join_check_params(setup);
+	case WD_GATHER:
+		return wd_gather_check_params(setup);
+	case WD_JOIN_GATHER:
+		ret = wd_join_check_params(setup);
+		if (ret)
+			return ret;
+
+		return wd_gather_check_params(setup);
+	default:
+		WD_ERR("invalid: hashjoin sess setup alg is wrong!\n");
+		return -WD_EINVAL;
+	}
+}
+
+static void sess_data_size_uninit(struct wd_join_gather_sess *sess)
+{
+	__u32 i;
+
+	if (sess->join_conf.cols)
+		free(sess->join_conf.cols);
+
+	if (sess->gather_conf.tables) {
+		for (i = 0; i < sess->gather_conf.table_num; i++)
+			free(sess->gather_conf.data_size[i]);
+
+		free(sess->gather_conf.tables);
+	}
+}
+
+static int sess_data_size_init(struct wd_join_gather_sess *sess,
+			       struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_gather_table_info *gtable = setup->gather_tables;
+	struct wd_join_table_info *jtable = &setup->join_table;
+	struct wd_join_gather_col_info *key = jtable->build_key_cols;
+	__u64 key_size, key_data_size, gather_size, gather_data_size;
+	__u32 i, j;
+
+	__atomic_store_n(&sess->state, WD_JOIN_SESS_UNINIT, __ATOMIC_RELEASE);
+
+	if (setup->alg != WD_GATHER) {
+		key_size = jtable->build_key_cols_num * sizeof(struct wd_join_gather_col_info);
+		key_data_size = jtable->build_key_cols_num * sizeof(__u64);
+		sess->join_conf.cols = malloc(key_size + key_data_size);
+		if (!sess->join_conf.cols)
+			return -WD_ENOMEM;
+		memcpy(sess->join_conf.cols, key, key_size);
+
+		sess->join_conf.data_size = (void *)sess->join_conf.cols + key_size;
+		for (i = 0; i < jtable->build_key_cols_num; i++)
+			(void)get_data_type_size(key[i].data_type, key[i].data_info,
+						 sess->join_conf.data_size, i);
+		sess->join_conf.cols_num = jtable->build_key_cols_num;
+
+		if (setup->alg == WD_JOIN)
+			return WD_SUCCESS;
+	}
+
+	gather_size = setup->gather_table_num * sizeof(struct wd_gather_table_info);
+	gather_data_size = setup->gather_table_num * sizeof(__u64 *);
+	sess->gather_conf.tables = malloc(gather_size + gather_data_size);
+	if (!sess->gather_conf.tables)
+		goto free_join;
+	memcpy(sess->gather_conf.tables, gtable, gather_size);
+
+	sess->gather_conf.data_size = (void *)sess->gather_conf.tables + gather_size;
+	for (i = 0; i < setup->gather_table_num; i++) {
+		sess->gather_conf.data_size[i] = malloc(gtable[i].cols_num * sizeof(__u64));
+		if (!sess->gather_conf.data_size[i])
+			goto free_gather;
+	}
+
+	for (i = 0; i < setup->gather_table_num; i++)
+		for (j = 0; j < gtable[i].cols_num; j++)
+			(void)get_data_type_size(gtable[i].cols[j].data_type,
+						 gtable[i].cols[j].data_info,
+						 sess->gather_conf.data_size[i], j);
+	sess->gather_conf.table_num = setup->gather_table_num;
+
+	return WD_SUCCESS;
+
+free_gather:
+	for (j = 0; j < i; j++)
+		free(sess->gather_conf.data_size[j]);
+	free(sess->gather_conf.tables);
+free_join:
+	if (setup->alg != WD_GATHER)
+		free(sess->join_conf.cols);
+	return -WD_ENOMEM;
+}
+
+static void wd_join_gather_uninit_sess(struct wd_join_gather_sess *sess)
+{
+	if (sess->gather_conf.batch_row_size)
+		free(sess->gather_conf.batch_row_size);
+
+	if (sess->ops.sess_uninit)
+		sess->ops.sess_uninit(wd_join_gather_setting.driver, sess->priv);
+}
+
+static int wd_join_gather_init_sess(struct wd_join_gather_sess *sess,
+				    struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_alg_driver *drv = wd_join_gather_setting.driver;
+	__u32 array_size;
+	int ret;
+
+	if (sess->ops.sess_init) {
+		if (!sess->ops.sess_uninit) {
+			WD_ERR("failed to get session uninit ops!\n");
+			return -WD_EINVAL;
+		}
+		ret = sess->ops.sess_init(drv, setup, &sess->priv);
+		if (ret) {
+			WD_ERR("failed to init session priv!\n");
+			return ret;
+		}
+	}
+
+	if (sess->ops.get_table_row_size && setup->alg != WD_GATHER) {
+		ret = sess->ops.get_table_row_size(drv, sess->priv);
+		if (ret <= 0) {
+			WD_ERR("failed to get hash table row size: %d!\n", ret);
+			goto uninit;
+		}
+		sess->hash_table.table_row_size = ret;
+	}
+
+	if (sess->ops.get_batch_row_size && setup->alg != WD_JOIN) {
+		array_size = setup->gather_table_num * sizeof(__u32);
+		sess->gather_conf.batch_row_size = malloc(array_size);
+		if (!sess->gather_conf.batch_row_size)
+			goto uninit;
+
+		ret = sess->ops.get_batch_row_size(drv, sess->priv,
+						   sess->gather_conf.batch_row_size,
+						   array_size);
+		if (ret) {
+			WD_ERR("failed to get batch table row size!\n");
+			goto free_batch;
+		}
+	}
+
+	return WD_SUCCESS;
+
+free_batch:
+	free(sess->gather_conf.batch_row_size);
+uninit:
+	if (sess->ops.sess_uninit)
+		sess->ops.sess_uninit(drv, sess->priv);
+	return -WD_EINVAL;
+}
+
+handle_t wd_join_gather_alloc_sess(struct wd_join_gather_sess_setup *setup)
+{
+	struct wd_join_gather_sess *sess;
+	int ret;
+
+	ret = wd_join_gather_check_params(setup);
+	if (ret)
+		return (handle_t)0;
+
+	sess = malloc(sizeof(struct wd_join_gather_sess));
+	if (!sess) {
+		WD_ERR("failed to alloc join gather session memory!\n");
+		return (handle_t)0;
+	}
+	memset(sess, 0, sizeof(struct wd_join_gather_sess));
+
+	sess->alg = setup->alg;
+	sess->index_type = setup->index_type;
+	sess->join_conf.key_output_enable = setup->join_table.key_output_enable;
+
+	ret = wd_drv_alg_support(wd_join_gather_alg[sess->alg], wd_join_gather_setting.driver);
+	if (!ret) {
+		WD_ERR("failed to check driver alg: %s!\n", wd_join_gather_alg[sess->alg]);
+		goto free_sess;
+	}
+
+	/* Some simple scheduler don't need scheduling parameters */
+	sess->sched_key = (void *)wd_join_gather_setting.sched.sched_init(
+		wd_join_gather_setting.sched.h_sched_ctx, setup->sched_param);
+	if (WD_IS_ERR(sess->sched_key)) {
+		WD_ERR("failed to init join_gather session schedule key!\n");
+		goto free_sess;
+	}
+
+	if (wd_join_gather_setting.driver->get_extend_ops) {
+		ret = wd_join_gather_setting.driver->get_extend_ops(&sess->ops);
+		if (ret) {
+			WD_ERR("failed to get join gather extend ops!\n");
+			goto free_key;
+		}
+	}
+
+	ret = wd_join_gather_init_sess(sess, setup);
+	if (ret)
+		goto free_key;
+
+	ret = sess_data_size_init(sess, setup);
+	if (ret) {
+		WD_ERR("failed to init join gather session data size!\n");
+		goto uninit_sess;
+	}
+
+	return (handle_t)sess;
+
+uninit_sess:
+	wd_join_gather_uninit_sess(sess);
+free_key:
+	free(sess->sched_key);
+free_sess:
+	free(sess);
+	return (handle_t)0;
+}
+
+void wd_join_gather_free_sess(handle_t h_sess)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+
+	if (!sess) {
+		WD_ERR("invalid: join gather input sess is NULL!\n");
+		return;
+	}
+
+	sess_data_size_uninit(sess);
+
+	wd_join_gather_uninit_sess(sess);
+
+	if (sess->sched_key)
+		free(sess->sched_key);
+
+	free(sess);
+}
+
+int wd_gather_get_batch_rowsize(handle_t h_sess, __u8 table_index)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+
+	if (!sess || !sess->gather_conf.batch_row_size) {
+		WD_ERR("invalid: gather input sess is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (table_index >= sess->gather_conf.table_num) {
+		WD_ERR("invalid: gather input sess is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	return sess->gather_conf.batch_row_size[table_index];
+}
+
+int wd_join_get_table_rowsize(handle_t h_sess)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+
+	if (!sess) {
+		WD_ERR("invalid: hashjoin input sess is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (sess->alg != WD_JOIN && sess->alg != WD_JOIN_GATHER) {
+		WD_ERR("invalid: the session is not used for hashjoin!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!sess->hash_table.table_row_size) {
+		WD_ERR("invalid:  hashjoin sess hash table row size is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	return sess->hash_table.table_row_size;
+}
+
+static int wd_join_init_sess_state(struct wd_join_gather_sess *sess,
+				   enum wd_join_sess_state *expected)
+{
+	enum wd_join_sess_state next;
+	int ret;
+
+	if (sess->hash_table.std_table) {
+		*expected = WD_JOIN_SESS_BUILD_HASH;
+		next = WD_JOIN_SESS_PREPARE_REHASH;
+	} else {
+		*expected = WD_JOIN_SESS_UNINIT;
+		next = WD_JOIN_SESS_INIT;
+	}
+
+	ret = __atomic_compare_exchange_n(&sess->state, expected, next,
+					  false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
+	if (!ret) {
+		WD_ERR("invalid: join sess state is %u!\n", *expected);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+int wd_join_set_hash_table(handle_t h_sess, struct wd_dae_hash_table *info)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected;
+	int ret;
+
+	if (!sess || !info) {
+		WD_ERR("invalid: hashjoin sess or hash table is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (sess->alg != WD_JOIN && sess->alg != WD_JOIN_GATHER) {
+		WD_ERR("invalid: the session is not used for hashjoin!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = wd_join_init_sess_state(sess, &expected);
+	if (ret)
+		return ret;
+
+	if (info->table_row_size != sess->hash_table.table_row_size) {
+		WD_ERR("invalid: hash table row size is not equal, expt: %u, real: %u!\n",
+		       sess->hash_table.table_row_size, info->table_row_size);
+		ret = -WD_EINVAL;
+		goto out;
+	}
+
+	if (!info->std_table) {
+		WD_ERR("invalid: standard hash table is NULL!\n");
+		ret = -WD_EINVAL;
+		goto out;
+	}
+
+	if (info->std_table_row_num < sess->hash_table.std_table_row_num) {
+		WD_ERR("invalid: standard hash table is too small, expt: %u, real: %u!\n",
+		       sess->hash_table.std_table_row_num, info->std_table_row_num);
+		ret = -WD_EINVAL;
+		goto out;
+	}
+
+	if (!info->ext_table_row_num || !info->ext_table)
+		WD_INFO("info: extern hash table is NULL!\n");
+
+	if (sess->ops.hash_table_init) {
+		ret = sess->ops.hash_table_init(wd_join_gather_setting.driver,
+						info, sess->priv);
+		if (ret)
+			goto out;
+	}
+
+	memcpy(&sess->hash_table, info, sizeof(struct wd_dae_hash_table));
+
+	return WD_SUCCESS;
+
+out:
+	__atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE);
+	return ret;
+}
+
+static void wd_join_gather_clear_status(void)
+{
+	wd_alg_clear_init(&wd_join_gather_setting.status);
+}
+
+static int wd_join_gather_alg_init(struct wd_ctx_config *config, struct wd_sched *sched)
+{
+	int ret;
+
+	ret = wd_set_epoll_en("WD_JOIN_GATHER_EPOLL_EN", &wd_join_gather_setting.config.epoll_en);
+	if (ret < 0)
+		return ret;
+
+	ret = wd_init_ctx_config(&wd_join_gather_setting.config, config);
+	if (ret < 0)
+		return ret;
+
+	ret = wd_init_sched(&wd_join_gather_setting.sched, sched);
+	if (ret < 0)
+		goto out_clear_ctx_config;
+
+	/* Allocate async pool for every ctx */
+	ret = wd_init_async_request_pool(&wd_join_gather_setting.pool, config, WD_POOL_MAX_ENTRIES,
+					 sizeof(struct wd_join_gather_msg));
+	if (ret < 0)
+		goto out_clear_sched;
+
+	ret = wd_alg_init_driver(&wd_join_gather_setting.config, wd_join_gather_setting.driver);
+	if (ret)
+		goto out_clear_pool;
+
+	return WD_SUCCESS;
+
+out_clear_pool:
+	wd_uninit_async_request_pool(&wd_join_gather_setting.pool);
+out_clear_sched:
+	wd_clear_sched(&wd_join_gather_setting.sched);
+out_clear_ctx_config:
+	wd_clear_ctx_config(&wd_join_gather_setting.config);
+	return ret;
+}
+
+static int wd_join_gather_alg_uninit(void)
+{
+	enum wd_status status;
+
+	wd_alg_get_init(&wd_join_gather_setting.status, &status);
+	if (status == WD_UNINIT)
+		return -WD_EINVAL;
+
+	/* Uninit async request pool */
+	wd_uninit_async_request_pool(&wd_join_gather_setting.pool);
+
+	/* Unset config, sched, driver */
+	wd_clear_sched(&wd_join_gather_setting.sched);
+
+	wd_alg_uninit_driver(&wd_join_gather_setting.config, wd_join_gather_setting.driver);
+
+	return WD_SUCCESS;
+}
+
+int wd_join_gather_init(char *alg, __u32 sched_type, int task_type,
+			struct wd_ctx_params *ctx_params)
+{
+	struct wd_ctx_params join_gather_ctx_params = {0};
+	struct wd_ctx_nums join_gather_ctx_num = {0};
+	int ret = -WD_EINVAL;
+	int state;
+	bool flag;
+
+	pthread_atfork(NULL, NULL, wd_join_gather_clear_status);
+
+	state = wd_alg_try_init(&wd_join_gather_setting.status);
+	if (state)
+		return state;
+
+	if (!alg || sched_type >= SCHED_POLICY_BUTT ||
+	    task_type < 0 || task_type >= TASK_MAX_TYPE) {
+		WD_ERR("invalid: join_gathe init input param is wrong!\n");
+		goto out_uninit;
+	}
+
+	flag = wd_join_gather_alg_check(alg);
+	if (!flag) {
+		WD_ERR("invalid: alg: %s is unsupported!\n", alg);
+		goto out_uninit;
+	}
+
+	state = wd_join_gather_open_driver();
+	if (state)
+		goto out_uninit;
+
+	while (ret != 0) {
+		memset(&wd_join_gather_setting.config, 0, sizeof(struct wd_ctx_config_internal));
+
+		/* Get alg driver and dev name */
+		wd_join_gather_setting.driver = wd_alg_drv_bind(task_type, alg);
+		if (!wd_join_gather_setting.driver) {
+			WD_ERR("failed to bind %s driver.\n", alg);
+			goto out_dlopen;
+		}
+
+		join_gather_ctx_params.ctx_set_num = &join_gather_ctx_num;
+		ret = wd_ctx_param_init(&join_gather_ctx_params, ctx_params,
+					wd_join_gather_setting.driver,
+					WD_JOIN_GATHER_TYPE, 1);
+		if (ret) {
+			if (ret == -WD_EAGAIN) {
+				wd_disable_drv(wd_join_gather_setting.driver);
+				wd_alg_drv_unbind(wd_join_gather_setting.driver);
+				continue;
+			}
+			goto out_driver;
+		}
+
+		(void)strcpy(wd_join_gather_init_attrs.alg, alg);
+		wd_join_gather_init_attrs.sched_type = sched_type;
+		wd_join_gather_init_attrs.driver = wd_join_gather_setting.driver;
+		wd_join_gather_init_attrs.ctx_params = &join_gather_ctx_params;
+		wd_join_gather_init_attrs.alg_init = wd_join_gather_alg_init;
+		wd_join_gather_init_attrs.alg_poll_ctx = wd_join_gather_poll_ctx;
+		ret = wd_alg_attrs_init(&wd_join_gather_init_attrs);
+		if (ret) {
+			if (ret == -WD_ENODEV) {
+				wd_disable_drv(wd_join_gather_setting.driver);
+				wd_alg_drv_unbind(wd_join_gather_setting.driver);
+				wd_ctx_param_uninit(&join_gather_ctx_params);
+				continue;
+			}
+			WD_ERR("fail to init alg attrs.\n");
+			goto out_params_uninit;
+		}
+	}
+
+	wd_alg_set_init(&wd_join_gather_setting.status);
+	wd_ctx_param_uninit(&join_gather_ctx_params);
+
+	return WD_SUCCESS;
+
+out_params_uninit:
+	wd_ctx_param_uninit(&join_gather_ctx_params);
+out_driver:
+	wd_alg_drv_unbind(wd_join_gather_setting.driver);
+out_dlopen:
+	wd_join_gather_close_driver();
+out_uninit:
+	wd_alg_clear_init(&wd_join_gather_setting.status);
+	return ret;
+}
+
+void wd_join_gather_uninit(void)
+{
+	int ret;
+
+	ret = wd_join_gather_alg_uninit();
+	if (ret)
+		return;
+
+	wd_alg_attrs_uninit(&wd_join_gather_init_attrs);
+	wd_alg_drv_unbind(wd_join_gather_setting.driver);
+	wd_join_gather_close_driver();
+	wd_join_gather_setting.dlh_list = NULL;
+	wd_alg_clear_init(&wd_join_gather_setting.status);
+}
+
+static void fill_build_hash_msg(struct wd_join_gather_msg *msg,
+				struct wd_join_gather_sess *sess)
+{
+	msg->index_type = sess->index_type;
+	msg->key_cols_num = sess->join_conf.cols_num;
+}
+
+static void fill_probe_msg(struct wd_join_gather_msg *msg,
+			   struct wd_join_gather_sess *sess)
+{
+	msg->key_cols_num = sess->join_conf.cols_num;
+	msg->index_type = sess->index_type;
+	msg->key_out_en = sess->join_conf.key_output_enable;
+}
+
+static void fill_rehash_msg(struct wd_join_gather_msg *msg,
+			    struct wd_join_gather_sess *sess)
+{
+	msg->key_cols_num = sess->join_conf.cols_num;
+}
+
+static void fill_complete_msg(struct wd_join_gather_msg *msg,
+			      struct wd_join_gather_sess *sess)
+{
+	__u32 table_index = msg->req.gather_req.table_index;
+
+	msg->index_type = sess->index_type;
+	msg->multi_batch_en = sess->gather_conf.tables[table_index].is_multi_batch;
+}
+
+
+static void fill_join_gather_msg(struct wd_join_gather_msg *msg, struct wd_join_gather_req *req,
+				 struct wd_join_gather_sess *sess)
+{
+	memcpy(&msg->req, req, sizeof(struct wd_join_gather_req));
+	msg->priv = sess->priv;
+	msg->op_type = req->op_type;
+
+	switch (req->op_type) {
+	case WD_JOIN_BUILD_HASH:
+		fill_build_hash_msg(msg, sess);
+		break;
+	case WD_JOIN_PROBE:
+		fill_probe_msg(msg, sess);
+		break;
+	case WD_JOIN_REHASH:
+		fill_rehash_msg(msg, sess);
+		break;
+	case WD_GATHER_CONVERT:
+		break;
+	case WD_GATHER_COMPLETE:
+		fill_complete_msg(msg, sess);
+		break;
+	default:
+		break;
+	}
+}
+
+static int wd_join_gather_check_common(struct wd_join_gather_sess *sess,
+				       struct wd_join_gather_req *req,
+				       __u8 mode, bool is_join)
+{
+	if (!sess) {
+		WD_ERR("invalid: join or gather session is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!req) {
+		WD_ERR("invalid: join input req is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (mode == CTX_MODE_ASYNC && !req->cb) {
+		WD_ERR("invalid: join gather req cb is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	switch (sess->alg) {
+	case WD_JOIN:
+		if (!is_join || !sess->join_conf.data_size) {
+			WD_ERR("invalid: join session data size is NULL!\n");
+			return -WD_EINVAL;
+		}
+		break;
+	case WD_GATHER:
+		if (is_join || !sess->gather_conf.data_size) {
+			WD_ERR("invalid: gather session data size is NULL!\n");
+			return -WD_EINVAL;
+		}
+		break;
+	case WD_JOIN_GATHER:
+		if (mode == CTX_MODE_ASYNC) {
+			WD_ERR("join-gather session does not support the async mode!\n");
+			return -WD_EINVAL;
+		}
+
+		if (!sess->join_conf.data_size || !sess->gather_conf.data_size) {
+			WD_ERR("invalid: join or gather session data size is NULL!\n");
+			return -WD_EINVAL;
+		}
+		break;
+	default:
+		WD_ERR("invalid: session alg is not supported!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int check_in_col_addr(struct wd_dae_col_addr *col, __u32 row_count,
+			     enum wd_dae_data_type type, __u64 data_size)
+{
+	if (!col->empty || col->empty_size != row_count * sizeof(col->empty[0])) {
+		WD_ERR("failed to check input empty col, size: %llu!\n", col->empty_size);
+		return -WD_EINVAL;
+	}
+
+	if (!col->value || col->value_size != row_count * data_size) {
+		WD_ERR("failed to check input value col size: %llu!\n", col->value_size);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int check_out_col_addr(struct wd_dae_col_addr *col, __u32 row_count,
+			      enum wd_dae_data_type type, __u64 data_size)
+{
+	if (!col->empty || col->empty_size < row_count * sizeof(col->empty[0])) {
+		WD_ERR("failed to check output empty col, size: %llu!\n", col->empty_size);
+		return -WD_EINVAL;
+	}
+
+	if (!col->value || col->value_size < row_count * data_size) {
+		WD_ERR("failed to check output value col size: %llu!\n", col->value_size);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int check_key_col_addr(struct wd_dae_col_addr *cols, __u32 cols_num,
+			      struct wd_join_gather_sess *sess, __u32 row_count, bool is_input)
+{
+	int (*func)(struct wd_dae_col_addr *col, __u32 row_count,
+		    enum wd_dae_data_type type, __u64 data_size);
+	__u32 i;
+	int ret;
+
+	func = is_input ? check_in_col_addr : check_out_col_addr;
+
+	for (i = 0; i < cols_num; i++) {
+		ret = func(cols + i, row_count, sess->join_conf.cols[i].data_type,
+			   sess->join_conf.data_size[i]);
+		if (ret) {
+			WD_ERR("failed to check req key col! col idx: %u\n", i);
+			return ret;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int check_data_col_addr(struct wd_gather_req *req, struct wd_join_gather_sess *sess,
+			       __u32 row_count, bool is_input)
+{
+	struct wd_gather_table_info *table = &sess->gather_conf.tables[req->table_index];
+	__u64 *data_size = sess->gather_conf.data_size[req->table_index];
+	int (*func)(struct wd_dae_col_addr *col, __u32 row_count,
+		    enum wd_dae_data_type type, __u64 data_size);
+	__u32 i;
+	int ret;
+
+	if (!data_size) {
+		WD_ERR("invalid: gather session data size is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!row_count) {
+		WD_ERR("invalid: gather data row number is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	func = is_input ? check_in_col_addr : check_out_col_addr;
+
+	for (i = 0; i < req->data_cols_num; i++) {
+		ret = func(&req->data_cols[i], row_count, table->cols[i].data_type,
+			   data_size[i]);
+		if (ret) {
+			WD_ERR("failed to check req data col! col idx: %u\n", i);
+			return ret;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int check_probe_out_addr(struct wd_probe_out_info *output,
+				struct wd_join_gather_sess *sess, __u32 row_num)
+{
+	if (!output->build_index.addr || !output->build_index.row_size) {
+		WD_ERR("probe multi index is not set!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!output->probe_index.addr || !output->probe_index.row_size) {
+		WD_ERR("probe single index is not set!\n");
+		return -WD_EINVAL;
+	}
+
+	if (output->build_index.row_num < row_num || output->probe_index.row_num < row_num) {
+		WD_ERR("build: %u, probe: %u, row num is less than output row_num: %u!\n",
+		       output->build_index.row_num, output->probe_index.row_num, row_num);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_common_check_req(struct wd_join_gather_sess *sess,
+				    struct wd_join_gather_req *req)
+{
+	struct wd_join_req *join_req = &req->join_req;
+	int ret;
+
+	if (join_req->key_cols_num != sess->join_conf.cols_num) {
+		WD_ERR("invalid: join table key_cols_num is not equal!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!join_req->key_cols) {
+		WD_ERR("invalid: join table key_cols is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!req->input_row_num) {
+		WD_ERR("invalid: join table input row number is zero!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = check_key_col_addr(join_req->key_cols, join_req->key_cols_num, sess,
+				 req->input_row_num, true);
+	if (ret) {
+		WD_ERR("failed to check join table key cols addr!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_build_hash_check_params(struct wd_join_gather_sess *sess,
+				      struct wd_join_gather_req *req, __u8 mode)
+{
+	int ret;
+
+	ret = wd_join_gather_check_common(sess, req, mode, true);
+	if (ret)
+		return ret;
+
+	if (req->op_type != WD_JOIN_BUILD_HASH) {
+		WD_ERR("failed to check req op_type for build hash task!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = wd_join_common_check_req(sess, req);
+	if (ret)
+		WD_ERR("failed to check join req for build hash task!\n");
+
+	return ret;
+}
+
+static int wd_join_probe_check_req(struct wd_join_gather_sess *sess,
+				   struct wd_join_gather_req *req)
+{
+	struct wd_join_req *jreq = &req->join_req;
+	struct wd_probe_out_info *probe_output = &jreq->probe_output;
+	int ret;
+
+	if (req->op_type != WD_JOIN_PROBE) {
+		WD_ERR("failed to check req op_type for probe task!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = wd_join_common_check_req(sess, req);
+	if (ret) {
+		WD_ERR("failed to check join req for probe task!\n");
+		return ret;
+	}
+
+	if (!req->output_row_num) {
+		WD_ERR("probe output row number is zero!\n");
+		return -WD_EINVAL;
+	}
+
+	if (sess->join_conf.key_output_enable) {
+		if (probe_output->key_cols_num != sess->join_conf.cols_num ||
+		    !probe_output->key_cols) {
+			WD_ERR("invalid: probe out key_cols_num is not equal!\n");
+			return -WD_EINVAL;
+		}
+		ret = check_key_col_addr(probe_output->key_cols, probe_output->key_cols_num,
+					 sess, req->output_row_num, false);
+		if (ret) {
+			WD_ERR("failed to check porbe output key cols addr!\n");
+			return -WD_EINVAL;
+		}
+	}
+
+	ret = check_probe_out_addr(probe_output, sess, req->output_row_num);
+	if (ret) {
+		WD_ERR("failed to check porbe output addr!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_probe_check_params(struct wd_join_gather_sess *sess,
+				      struct wd_join_gather_req *req, __u8 mode)
+{
+	int ret;
+
+	ret = wd_join_gather_check_common(sess, req, mode, true);
+	if (ret)
+		return ret;
+
+	return wd_join_probe_check_req(sess, req);
+}
+
+static int wd_join_rehash_check_params(struct wd_join_gather_sess *sess,
+				       struct wd_join_gather_req *req)
+{
+	int ret;
+
+	ret = wd_join_gather_check_common(sess, req, CTX_MODE_SYNC, true);
+	if (ret)
+		return ret;
+
+	if (req->op_type != WD_JOIN_REHASH) {
+		WD_ERR("failed to check req op_type for rehash task!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!req->output_row_num) {
+		WD_ERR("invalid: req output_row_num is 0 for join rehash!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_gather_sync_job(struct wd_join_gather_sess *sess,
+				   struct wd_join_gather_req *req,
+				   struct wd_join_gather_msg *msg)
+{
+	struct wd_join_gather_setting *setting = &wd_join_gather_setting;
+	struct wd_ctx_config_internal *config = &setting->config;
+	struct wd_msg_handle msg_handle;
+	struct wd_ctx_internal *ctx;
+	__u32 idx;
+	int ret;
+
+	memset(msg, 0, sizeof(struct wd_join_gather_msg));
+	fill_join_gather_msg(msg, req, sess);
+	req->state = 0;
+
+	idx = setting->sched.pick_next_ctx(setting->sched.h_sched_ctx,
+					   sess->sched_key, CTX_MODE_SYNC);
+	ret = wd_check_ctx(config, CTX_MODE_SYNC, idx);
+	if (ret)
+		return ret;
+
+	wd_dfx_msg_cnt(config, WD_CTX_CNT_NUM, idx);
+	ctx = config->ctxs + idx;
+
+	msg_handle.send = setting->driver->send;
+	msg_handle.recv = setting->driver->recv;
+
+	pthread_spin_lock(&ctx->lock);
+	ret = wd_handle_msg_sync(setting->driver, &msg_handle, ctx->ctx,
+				 msg, NULL, config->epoll_en);
+	pthread_spin_unlock(&ctx->lock);
+
+	return ret;
+}
+
+static int wd_build_hash_try_init(struct wd_join_gather_sess *sess,
+				  enum wd_join_sess_state *expected)
+{
+	enum wd_join_sess_state state;
+
+	(void)__atomic_compare_exchange_n(&sess->state, expected, WD_JOIN_SESS_BUILD_HASH,
+					  false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
+	state = __atomic_load_n(&sess->state, __ATOMIC_RELAXED);
+	if (state != WD_JOIN_SESS_BUILD_HASH) {
+		WD_ERR("failed to set join sess state: %u!\n", state);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_gather_check_result(__u32 result)
+{
+	switch (result) {
+	case WD_JOIN_GATHER_TASK_DONE:
+		return WD_SUCCESS;
+	case WD_JOIN_GATHER_IN_EPARA:
+	case WD_JOIN_GATHER_NEED_REHASH:
+	case WD_JOIN_GATHER_INVALID_HASH_TABLE:
+	case WD_JOIN_GATHER_PARSE_ERROR:
+	case WD_JOIN_GATHER_BUS_ERROR:
+		WD_ERR("failed to check join gather message state: %u!\n", result);
+		return -WD_EIO;
+	default:
+		return -WD_EINVAL;
+	}
+}
+
+int wd_join_build_hash_sync(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected = WD_JOIN_SESS_INIT;
+	struct wd_join_gather_msg msg;
+	int ret;
+
+	ret = wd_build_hash_check_params(sess, req, CTX_MODE_SYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check hashjoin build hash params!\n");
+		return ret;
+	}
+
+	ret = wd_build_hash_try_init(sess, &expected);
+	if (unlikely(ret))
+		return ret;
+
+	ret = wd_join_gather_sync_job(sess, req, &msg);
+	if (unlikely(ret)) {
+		if (expected == WD_JOIN_SESS_INIT)
+			__atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE);
+		WD_ERR("failed to do hashjoin build hash sync job!\n");
+		return ret;
+	}
+
+	req->consumed_row_num = msg.consumed_row_num;
+	req->state = msg.result;
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_gather_async_job(struct wd_join_gather_sess *sess,
+				    struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_setting *setting = &wd_join_gather_setting;
+	struct wd_ctx_config_internal *config = &setting->config;
+	struct wd_join_gather_msg *msg;
+	struct wd_ctx_internal *ctx;
+	int msg_id, ret;
+	__u32 idx;
+
+	idx = setting->sched.pick_next_ctx(setting->sched.h_sched_ctx,
+					   sess->sched_key, CTX_MODE_ASYNC);
+	ret = wd_check_ctx(config, CTX_MODE_ASYNC, idx);
+	if (ret)
+		return ret;
+
+	ctx = config->ctxs + idx;
+	msg_id = wd_get_msg_from_pool(&setting->pool, idx, (void **)&msg);
+	if (msg_id < 0) {
+		WD_ERR("failed to get join gather msg from pool!\n");
+		return msg_id;
+	}
+
+	fill_join_gather_msg(msg, req, sess);
+	msg->tag = msg_id;
+	ret = wd_alg_driver_send(setting->driver, ctx->ctx, msg);
+	if (ret < 0) {
+		if (ret != -WD_EBUSY)
+			WD_ERR("wd join gather async send err!\n");
+
+		goto fail_with_msg;
+	}
+
+	wd_dfx_msg_cnt(config, WD_CTX_CNT_NUM, idx);
+
+	return WD_SUCCESS;
+
+fail_with_msg:
+	wd_put_msg_to_pool(&setting->pool, idx, msg->tag);
+	return ret;
+}
+
+int wd_join_build_hash_async(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected = WD_JOIN_SESS_INIT;
+	int ret;
+
+	ret = wd_build_hash_check_params(sess, req, CTX_MODE_ASYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check build hash async params!\n");
+		return ret;
+	}
+
+	ret = wd_build_hash_try_init(sess, &expected);
+	if (unlikely(ret))
+		return ret;
+
+	ret = wd_join_gather_async_job(sess, req);
+	if (unlikely(ret)) {
+		if (expected == WD_JOIN_SESS_INIT)
+			__atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE);
+		WD_ERR("failed to do join build hash async job!\n");
+	}
+
+	return ret;
+}
+
+static int wd_join_probe_try_init(struct wd_join_gather_sess *sess,
+				  enum wd_join_sess_state *expected)
+{
+	enum wd_join_sess_state state;
+
+	(void)__atomic_compare_exchange_n(&sess->state, expected, WD_JOIN_SESS_PROBE,
+					  false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
+	state = __atomic_load_n(&sess->state, __ATOMIC_RELAXED);
+	if (state != WD_JOIN_SESS_PROBE) {
+		WD_ERR("failed to set join sess state: %u!\n", state);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+int wd_join_probe_sync(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected = WD_JOIN_SESS_BUILD_HASH;
+	struct wd_join_gather_msg msg;
+	int ret;
+
+	ret = wd_join_probe_check_params(sess, req, CTX_MODE_SYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check join probe params!\n");
+		return ret;
+	}
+
+	ret = wd_join_probe_try_init(sess, &expected);
+	if (unlikely(ret))
+		return ret;
+
+	ret = wd_join_gather_sync_job(sess, req, &msg);
+	if (unlikely(ret)) {
+		if (expected == WD_JOIN_SESS_BUILD_HASH)
+			__atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE);
+		WD_ERR("failed to do join probe sync job!\n");
+		return ret;
+	}
+
+	req->consumed_row_num = msg.consumed_row_num;
+	req->produced_row_num = msg.produced_row_num;
+	req->output_done = msg.output_done;
+	req->state = msg.result;
+
+	return WD_SUCCESS;
+}
+
+int wd_join_probe_async(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected = WD_JOIN_SESS_BUILD_HASH;
+	int ret;
+
+	ret = wd_join_probe_check_params(sess, req, CTX_MODE_ASYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check join probe params!\n");
+		return ret;
+	}
+
+	ret = wd_join_probe_try_init(sess, &expected);
+	if (unlikely(ret))
+		return ret;
+
+	ret = wd_join_gather_async_job(sess, req);
+	if (unlikely(ret)) {
+		if (expected == WD_JOIN_SESS_BUILD_HASH)
+			__atomic_store_n(&sess->state, expected, __ATOMIC_RELEASE);
+		WD_ERR("failed to do join probe async job!\n");
+	}
+
+	return ret;
+}
+
+static int wd_join_rehash_sync_inner(struct wd_join_gather_sess *sess,
+				     struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_msg msg = {0};
+	int ret;
+
+	ret = wd_join_gather_sync_job(sess, req, &msg);
+	if (ret)
+		return ret;
+
+	ret = wd_join_gather_check_result(msg.result);
+	if (ret)
+		return ret;
+
+	req->output_done = msg.output_done;
+
+	return WD_SUCCESS;
+}
+
+static int wd_join_rehash_try_init(struct wd_join_gather_sess *sess,
+				   enum wd_join_sess_state *expected)
+{
+	int ret;
+
+	ret = __atomic_compare_exchange_n(&sess->state, expected, WD_JOIN_SESS_REHASH,
+					  false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
+	if (!ret) {
+		WD_ERR("invalid: join rehash sess state is %u!\n", *expected);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+int wd_join_rehash_sync(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	enum wd_join_sess_state expected = WD_JOIN_SESS_PREPARE_REHASH;
+	__u64 max_cnt, cnt = 0;
+	int ret;
+
+	ret = wd_join_rehash_check_params(sess, req);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check join rehash params!\n");
+		return ret;
+	}
+
+	ret = wd_join_rehash_try_init(sess, &expected);
+	if (unlikely(ret))
+		return ret;
+
+	max_cnt = MAX_HASH_TABLE_ROW_NUM / req->output_row_num;
+	while (cnt < max_cnt) {
+		ret = wd_join_rehash_sync_inner(sess, req);
+		if (unlikely(ret)) {
+			__atomic_store_n(&sess->state, WD_JOIN_SESS_PREPARE_REHASH,
+					 __ATOMIC_RELEASE);
+			WD_ERR("failed to do join rehash task!\n");
+			return ret;
+		}
+		if (req->output_done)
+			break;
+		cnt++;
+	}
+
+	__atomic_store_n(&sess->state, WD_JOIN_SESS_BUILD_HASH, __ATOMIC_RELEASE);
+	return WD_SUCCESS;
+}
+
+static int wd_gather_common_check_req(struct wd_join_gather_sess *sess,
+				      struct wd_join_gather_req *req)
+{
+	struct wd_gather_req *gather_req = &req->gather_req;
+	struct wd_gather_table_info *tables;
+	__u32 table_index;
+
+	if (!sess->gather_conf.tables) {
+		WD_ERR("invalid: session gather tables is NULL!\n");
+		return -WD_EINVAL;
+	}
+	tables = sess->gather_conf.tables;
+	table_index = gather_req->table_index;
+
+	if (table_index >= sess->gather_conf.table_num) {
+		WD_ERR("invalid: gather table index(%u) is too big!\n", table_index);
+		return -WD_EINVAL;
+	}
+
+	if (gather_req->data_cols_num != tables[table_index].cols_num) {
+		WD_ERR("invalid: gather table data_cols_num is not equal!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!gather_req->data_cols) {
+		WD_ERR("invalid: gather table data_cols is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_gather_convert_check_req(struct wd_join_gather_sess *sess,
+				       struct wd_join_gather_req *req)
+{
+	struct wd_gather_req *gather_req = &req->gather_req;
+	__u32 expt_size, table_index;
+	int ret;
+
+	if (req->op_type != WD_GATHER_CONVERT) {
+		WD_ERR("failed to check req op_type for gather convert task!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = wd_gather_common_check_req(sess, req);
+	if (ret)
+		return ret;
+
+	table_index = gather_req->table_index;
+
+	ret = check_data_col_addr(gather_req, sess, req->input_row_num, true);
+	if (ret) {
+		WD_ERR("failed to check gather convert data cols addr!\n");
+		return -WD_EINVAL;
+	}
+
+	if (gather_req->row_batchs.batch_num != 1 || !gather_req->row_batchs.batch_addr ||
+	    !gather_req->row_batchs.batch_addr[0]) {
+		WD_ERR("invalid: gather convert only support one batch!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!gather_req->row_batchs.batch_row_num || !gather_req->row_batchs.batch_row_size) {
+		WD_ERR("invalid: gather convert batchs row_num or row_size is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	expt_size = sess->gather_conf.batch_row_size[table_index];
+	if (gather_req->row_batchs.batch_row_num[0] != req->input_row_num ||
+	    gather_req->row_batchs.batch_row_size[0] != expt_size) {
+		WD_ERR("invalid: gather convert row batchs, row_size: %u, row_num: %u\n",
+		       gather_req->row_batchs.batch_row_size[0],
+		       gather_req->row_batchs.batch_row_num[0]);
+		return -WD_EINVAL;
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_gather_complete_check_req(struct wd_join_gather_sess *sess,
+					struct wd_join_gather_req *req)
+{
+	struct wd_gather_req *gather_req = &req->gather_req;
+	struct wd_gather_table_info *tables;
+	struct wd_dae_row_addr *index_addr;
+	__u32 table_index, expt_size, i;
+	int ret;
+
+	if (req->op_type != WD_GATHER_COMPLETE) {
+		WD_ERR("failed to check req op_type for gather complete task!\n");
+		return -WD_EINVAL;
+	}
+
+	ret = wd_gather_common_check_req(sess, req);
+	if (ret)
+		return ret;
+
+	tables = sess->gather_conf.tables;
+	table_index = gather_req->table_index;
+
+	ret = check_data_col_addr(gather_req, sess, req->output_row_num, false);
+	if (ret) {
+		WD_ERR("failed to check gather complete data cols addr!\n");
+		return -WD_EINVAL;
+	}
+
+	index_addr = &gather_req->index;
+	if (!index_addr->addr || index_addr->row_num < req->output_row_num) {
+		WD_ERR("invalid: gather index is NULL or index row number is small!\n");
+		return -WD_EINVAL;
+	}
+
+	/* The row batch information is stored to index, no need to check. */
+	if (sess->index_type == WD_BATCH_ADDR_INDEX && tables[table_index].is_multi_batch)
+		return WD_SUCCESS;
+
+	if (!gather_req->row_batchs.batch_num || !gather_req->row_batchs.batch_addr) {
+		WD_ERR("invalid: gather row batch is NULL or batch addr number is 0!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!gather_req->row_batchs.batch_row_num || !gather_req->row_batchs.batch_row_size) {
+		WD_ERR("invalid: gather row batch row_num or row_size is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	if (!tables[table_index].is_multi_batch) {
+		if (gather_req->row_batchs.batch_num != 1) {
+			WD_ERR("invalid: single gather row batch addr num should be 1!\n");
+			return -WD_EINVAL;
+		}
+	}
+
+	for (i = 0; i < gather_req->row_batchs.batch_num; i++) {
+		if (!gather_req->row_batchs.batch_addr[i] ||
+		    !gather_req->row_batchs.batch_row_num[i]) {
+			WD_ERR("invalid: row batch addr or row_num is null! idx: %u\n", i);
+			return -WD_EINVAL;
+		}
+		expt_size = sess->gather_conf.batch_row_size[table_index];
+		if (gather_req->row_batchs.batch_row_size[i] != expt_size) {
+			WD_ERR("invalid row batch row_size: %u, batch idx: %u\n",
+			       gather_req->row_batchs.batch_row_size[i], i);
+			return -WD_EINVAL;
+		}
+	}
+
+	return WD_SUCCESS;
+}
+
+static int wd_gather_convert_check_params(struct wd_join_gather_sess *sess,
+					  struct wd_join_gather_req *req, __u8 mode)
+{
+	int ret;
+
+	ret = wd_join_gather_check_common(sess, req, mode, false);
+	if (ret)
+		return ret;
+
+	return wd_gather_convert_check_req(sess, req);
+}
+
+static int wd_gather_complete_check_params(struct wd_join_gather_sess *sess,
+					   struct wd_join_gather_req *req, __u8 mode)
+{
+	int ret;
+
+	ret = wd_join_gather_check_common(sess, req, mode, false);
+	if (ret)
+		return ret;
+
+	return wd_gather_complete_check_req(sess, req);
+}
+
+int wd_gather_convert_sync(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	struct wd_join_gather_msg msg;
+	int ret;
+
+	ret = wd_gather_convert_check_params(sess, req, CTX_MODE_SYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check gather convert params!\n");
+		return ret;
+	}
+
+	ret = wd_join_gather_sync_job(sess, req, &msg);
+	if (unlikely(ret)) {
+		WD_ERR("failed to do gather convert sync job!\n");
+		return ret;
+	}
+
+	req->consumed_row_num = msg.consumed_row_num;
+	req->state = msg.result;
+
+	return WD_SUCCESS;
+}
+
+int wd_gather_convert_async(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	int ret;
+
+	ret = wd_gather_convert_check_params(sess, req, CTX_MODE_ASYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check gather convert async params!\n");
+		return ret;
+	}
+
+	ret = wd_join_gather_async_job(sess, req);
+	if (unlikely(ret))
+		WD_ERR("failed to do gather convert async job!\n");
+
+	return ret;
+}
+
+int wd_gather_complete_sync(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	struct wd_join_gather_msg msg;
+	int ret;
+
+	ret = wd_gather_complete_check_params(sess, req, CTX_MODE_SYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check gather complete params!\n");
+		return ret;
+	}
+
+	ret = wd_join_gather_sync_job(sess, req, &msg);
+	if (unlikely(ret)) {
+		WD_ERR("failed to do gather complete sync job!\n");
+		return ret;
+	}
+
+	req->produced_row_num = msg.produced_row_num;
+	req->state = msg.result;
+
+	return WD_SUCCESS;
+}
+
+int wd_gather_complete_async(handle_t h_sess, struct wd_join_gather_req *req)
+{
+	struct wd_join_gather_sess *sess = (struct wd_join_gather_sess *)h_sess;
+	int ret;
+
+	ret = wd_gather_complete_check_params(sess, req, CTX_MODE_ASYNC);
+	if (unlikely(ret)) {
+		WD_ERR("failed to check gather complete params!\n");
+		return ret;
+	}
+
+	ret = wd_join_gather_async_job(sess, req);
+	if (unlikely(ret))
+		WD_ERR("failed to do gather complete async job!\n");
+
+	return ret;
+}
+
+struct wd_join_gather_msg *wd_join_gather_get_msg(__u32 idx, __u32 tag)
+{
+	return wd_find_msg_in_pool(&wd_join_gather_setting.pool, idx, tag);
+}
+
+static int wd_join_gather_poll_ctx(__u32 idx, __u32 expt, __u32 *count)
+{
+	struct wd_ctx_config_internal *config = &wd_join_gather_setting.config;
+	struct wd_join_gather_msg resp_msg, *msg;
+	struct wd_ctx_internal *ctx;
+	struct wd_join_gather_req *req;
+	__u64 recv_count = 0;
+	__u32 tmp = expt;
+	int ret;
+
+	*count = 0;
+
+	ret = wd_check_ctx(config, CTX_MODE_ASYNC, idx);
+	if (unlikely(ret))
+		return ret;
+
+	ctx = config->ctxs + idx;
+
+	do {
+		ret = wd_alg_driver_recv(wd_join_gather_setting.driver, ctx->ctx, &resp_msg);
+		if (ret == -WD_EAGAIN) {
+			return ret;
+		} else if (ret < 0) {
+			WD_ERR("wd join_gather recv hw err!\n");
+			return ret;
+		}
+		recv_count++;
+		msg = wd_find_msg_in_pool(&wd_join_gather_setting.pool, idx, resp_msg.tag);
+		if (!msg) {
+			WD_ERR("failed to get join gather msg from pool!\n");
+			return -WD_EINVAL;
+		}
+
+		msg->req.state = resp_msg.result;
+		msg->req.consumed_row_num = resp_msg.consumed_row_num;
+		msg->req.produced_row_num = resp_msg.produced_row_num;
+		msg->req.output_done = resp_msg.output_done;
+		req = &msg->req;
+
+		req->cb(req, req->cb_param);
+		/* Free msg cache to msg_pool */
+		wd_put_msg_to_pool(&wd_join_gather_setting.pool, idx, resp_msg.tag);
+		*count = recv_count;
+	} while (--tmp);
+
+	return ret;
+}
+
+int wd_join_gather_poll(__u32 expt, __u32 *count)
+{
+	handle_t h_ctx = wd_join_gather_setting.sched.h_sched_ctx;
+	struct wd_sched *sched = &wd_join_gather_setting.sched;
+
+	if (!expt || !count) {
+		WD_ERR("invalid: join gather poll input param is NULL!\n");
+		return -WD_EINVAL;
+	}
+
+	return sched->poll_policy(h_ctx, expt, count);
+}
diff --git a/wd_util.c b/wd_util.c
index 646346a0..56d85d49 100644
--- a/wd_util.c
+++ b/wd_util.c
@@ -64,6 +64,7 @@ static const char *wd_env_name[WD_TYPE_MAX] = {
 	"WD_ECC_CTX_NUM",
 	"WD_AGG_CTX_NUM",
 	"WD_UDMA_CTX_NUM",
+	"WD_JOIN_GATHER_CTX_NUM",
 };
 
 struct async_task {
@@ -111,6 +112,9 @@ static struct acc_alg_item alg_options[] = {
 	{"lz77_only", "lz77_only"},
 	{"hashagg", "hashagg"},
 	{"udma", "udma"},
+	{"hashjoin", "hashjoin"},
+	{"gather", "gather"},
+	{"join-gather", "hashjoin"},
 
 	{"rsa", "rsa"},
 	{"dh", "dh"},
@@ -2594,7 +2598,7 @@ static int wd_alg_ctx_init(struct wd_init_attrs *attrs)
 
 	list = wd_get_accel_list(attrs->alg);
 	if (!list) {
-		WD_ERR("failed to get devices!\n");
+		WD_ERR("failed to get devices for alg: %s\n", attrs->alg);
 		return -WD_ENODEV;
 	}
 
-- 
2.33.0

    