From b488d43f568bebe5f6bb1081571d3967694f6a52 Mon Sep 17 00:00:00 2001
From: lilongbin <lilongbin3@huawei.com>
Date: Tue, 27 May 2025 11:17:12 +0800
Subject: [PATCH] ock adapter null flag

---
 .../cpp/src/shuffle/ock_merge_reader.cpp      | 19 +++++++++++--------
 .../cpp/src/shuffle/ock_merge_reader.h        |  2 +-
 .../ock-omniop-shuffle/pom.xml                |  6 +++---
 .../omniop-spark-extension-ock/pom.xml        |  6 +++---
 4 files changed, 18 insertions(+), 15 deletions(-)

diff --git a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.cpp b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.cpp
index d1ef824c4..6486af2c2 100644
--- a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.cpp
+++ b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.cpp
@@ -166,7 +166,7 @@ bool OckMergeReader::GetMergeVectorBatch(uint8_t *&startAddress, uint32_t remain
     return true;
 }
 
-bool OckMergeReader::CopyPartDataToVector(uint8_t *&nulls, uint8_t *&values, uint32_t &remainingSize, 
+bool OckMergeReader::CopyPartDataToVector(uint8_t *&nulls, uint32_t &nullsOffset, uint8_t *&values, uint32_t &remainingSize,
     uint32_t &remainingCapacity, OckVectorPtr &srcVector)
 {
     uint32_t srcSize = srcVector->GetSize();
@@ -174,12 +174,14 @@ bool OckMergeReader::CopyPartDataToVector(uint8_t *&nulls, uint8_t *&values, uin
         LOG_ERROR("Not eneough resource. remainingSize %d, srcSize %d.", remainingSize, srcSize);
         return false;
     }
-    errno_t ret = memcpy_s(nulls, remainingSize, srcVector->GetValueNulls(), srcSize);
-    if (UNLIKELY(ret != EOK)) {
-        LOG_ERROR("Failed to copy null vector");
-        return false;
+
+    uint8_t *srcNulls = reinterpret_cast<uint8_t *>(srcVector->GetValueNulls());
+    for (uint32_t i = 0; i < srcSize; i++) {
+        if (srcNulls[i]) {
+            omniruntime::BitUtil::SetBit(nulls, nullsOffset + i);
+        }
     }
-    nulls += srcSize;
+    nullsOffset += srcSize;
     remainingSize -= srcSize;
 
     uint32_t srcCapacity = srcVector->GetCapacityInBytes();
@@ -188,7 +190,7 @@ bool OckMergeReader::CopyPartDataToVector(uint8_t *&nulls, uint8_t *&values, uin
         return false;
     }
     if (srcCapacity > 0) {
-        ret = memcpy_s(values, remainingCapacity, srcVector->GetValues(), srcCapacity);
+        errno_t  ret = memcpy_s(values, remainingCapacity, srcVector->GetValues(), srcCapacity);
         if (UNLIKELY(ret != EOK)) {
             LOG_ERROR("Failed to copy values vector");
             return false;
@@ -224,13 +226,14 @@ bool OckMergeReader::CopyDataToVector(BaseVector *dstVector, uint32_t colIndex)
         remainingCapacity = GetDataSize(colIndex) * remainingSize;
     }
 
+    uint32_t nullsOffset = 0;
     for (uint32_t cnt = 0; cnt < mMergeCnt; ++cnt) {
         if (UNLIKELY(srcVector == nullptr)) {
             LOG_ERROR("Invalid src vector");
             return false;
         }
 
-        if (UNLIKELY(!CopyPartDataToVector(nullsAddress, valuesAddress, remainingSize, remainingCapacity, srcVector))) {
+        if (UNLIKELY(!CopyPartDataToVector(nullsAddress, nullsOffset, valuesAddress, remainingSize, remainingCapacity, srcVector))) {
             return false;
         }
 
diff --git a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.h b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.h
index 838dd6a8d..385d3711e 100644
--- a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.h
+++ b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/cpp/src/shuffle/ock_merge_reader.h
@@ -16,7 +16,7 @@ public:
     bool Initialize(const int32_t *typeIds, uint32_t colNum);
     bool GetMergeVectorBatch(uint8_t *&address, uint32_t remain, uint32_t maxRowNum, uint32_t maxSize);
 
-    bool CopyPartDataToVector(uint8_t *&nulls, uint8_t *&values, uint32_t &remainingSize, uint32_t &remainingCapacity,
+    bool CopyPartDataToVector(uint8_t *&nulls, uint32_t &nullsOffset, uint8_t *&values, uint32_t &remainingSize, uint32_t &remainingCapacity,
          OckVectorPtr &srcVector);
     bool CopyDataToVector(omniruntime::vec::BaseVector *dstVector, uint32_t colIndex);
 
diff --git a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/pom.xml b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/pom.xml
index 7457d49fe..92fbd8d5c 100644
--- a/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/pom.xml
+++ b/omnioperator/omniop-spark-extension-ock/ock-omniop-shuffle/pom.xml
@@ -6,7 +6,7 @@
    <parent>
        <groupId>com.huawei.ock</groupId>
        <artifactId>omniop-spark-extension-ock</artifactId>
-       <version>25.0.0</version>
+       <version>24.0.0</version>
    </parent>
    <properties>
      <cpp.dir>cpp/</cpp.dir>
@@ -18,7 +18,7 @@
    <dependencies>
      <dependency>
        <groupId>com.huawei.kunpeng</groupId>
-       <artifactId>boostkit-omniop-spark</artifactId>
+       <artifactId>spark-extension-core</artifactId>
        <version>3.3.1-1.9.0</version>
        <scope>compile</scope>
      </dependency>
@@ -27,7 +27,7 @@
    <artifactId>ock-omniop-shuffle-manager</artifactId>
    <packaging>jar</packaging>
    <description>Huawei Open Computing Kit for Spark, shuffle manager</description>
-   <version>25.0.0</version>
+   <version>24.0.0</version>
 
    <build>
      <!--suppress UnresolvedMavenProperty -->
diff --git a/omnioperator/omniop-spark-extension-ock/pom.xml b/omnioperator/omniop-spark-extension-ock/pom.xml
index ed8150677..6ea9000a4 100644
--- a/omnioperator/omniop-spark-extension-ock/pom.xml
+++ b/omnioperator/omniop-spark-extension-ock/pom.xml
@@ -8,7 +8,7 @@
    <artifactId>omniop-spark-extension-ock</artifactId>
    <packaging>pom</packaging>
    <description>Huawei Open Computing Kit for Spark</description>
-   <version>25.0.0</version>
+   <version>24.0.0</version>
 
    <properties>
      <spark.version>3.3.1</spark.version>
@@ -20,7 +20,7 @@
      <input.version>spark-3.3</input.version>
      <scala.plugin.version>3.2.0</scala.plugin.version>
      <maven.plugin.version>3.1.1</maven.plugin.version>
-     <global.version>25.0.0</global.version>
+     <global.version>24.0.0</global.version>
    </properties>
 
    <dependencies>
@@ -67,7 +67,7 @@
      </dependency>
      <dependency>
        <groupId>com.huawei.kunpeng</groupId>
-       <artifactId>boostkit-omniop-spark</artifactId>
+       <artifactId>spark-extension-core</artifactId>
        <version>3.3.1-1.9.0</version>
      </dependency>
      <dependency>
-- 
Gitee