fix: positional delete bug when base row number is greater than the largest

delete position

Author: nmahadevuni

velox/connectors/hive/iceberg/PositionalDeleteFileReader.cpp

@@ -223,6 +223,16 @@ void PositionalDeleteFileReader::updateDeleteBitmap(
       deletePositionsVector->as<FlatVector<int64_t>>()->rawValues();
   int64_t offset = baseReadOffset + splitOffset_;
 
+  // If the offset is greater than the last position in this delete rows batch,
+  // nothing to delete.
+  if (offset > deletePositions[deletePositionsVector->size() - 1]) {
+    return;
+  }
+
+  while (deletePositionsOffset_ < deletePositionsVector->size() &&
+         deletePositions[deletePositionsOffset_] < offset) {
+    deletePositionsOffset_++;
+  }
   while (deletePositionsOffset_ < deletePositionsVector->size() &&
          deletePositions[deletePositionsOffset_] < rowNumberUpperBound) {
     bits::setBit(

velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp

@@ -209,7 +209,8 @@ class HiveIcebergTest : public HiveConnectorTestBase {
         }
       }
 
-      splits.emplace_back(makeIcebergSplit(baseFilePath, deleteFiles));
+      auto icebergSplits = makeIcebergSplit(baseFilePath, 3, deleteFiles);
+      splits.insert(splits.end(), icebergSplits.begin(), icebergSplits.end());
     }
 
     std::string duckdbSql =
@@ -335,8 +336,9 @@ class HiveIcebergTest : public HiveConnectorTestBase {
     return vectors;
   }
 
-  std::shared_ptr<ConnectorSplit> makeIcebergSplit(
+  std::vector<std::shared_ptr<ConnectorSplit>> makeIcebergSplit(
       const std::string& dataFilePath,
+      const uint32_t splitCount = 1,
       const std::vector<IcebergDeleteFile>& deleteFiles = {}) {
     std::unordered_map<std::string, std::optional<std::string>> partitionKeys;
     std::unordered_map<std::string, std::string> customSplitInfo;
@@ -346,17 +348,24 @@ class HiveIcebergTest : public HiveConnectorTestBase {
                     ->openFileForRead(dataFilePath);
     const int64_t fileSize = file->size();
 
-    return std::make_shared<HiveIcebergSplit>(
-        kHiveConnectorId,
-        dataFilePath,
-        fileFomat_,
-        0,
-        fileSize,
-        partitionKeys,
-        std::nullopt,
-        customSplitInfo,
-        nullptr,
-        deleteFiles);
+    std::vector<std::shared_ptr<ConnectorSplit>> splits;
+    const uint64_t splitSize = std::ceil((fileSize) / splitCount);
+
+    for (int i = 0; i < splitCount; ++i) {
+      splits.emplace_back(std::make_shared<HiveIcebergSplit>(
+          kHiveConnectorId,
+          dataFilePath,
+          fileFomat_,
+          i * splitSize,
+          splitSize,
+          partitionKeys,
+          std::nullopt,
+          customSplitInfo,
+          nullptr,
+          deleteFiles));
+    }
+
+    return splits;
   }
 
   std::string getDuckDBQuery(
@@ -657,6 +666,8 @@ TEST_F(HiveIcebergTest, positionalDeletesMultipleSplits) {
   assertMultipleSplits(makeRandomIncreasingValues(0, 20000), 10, 3);
   assertMultipleSplits(makeContinuousIncreasingValues(0, 20000), 10, 3);
   assertMultipleSplits({}, 10, 3);
+
+  assertMultipleSplits(makeContinuousIncreasingValues(0, 5000), 1, 3);
 }
 
 } // namespace facebook::velox::connector::hive::iceberg