[VL][Delta] Fix Delta scan CI regressions

Author: Mohammad Linjawi

cpp/velox/compute/delta/DeltaDeletionVectorReader.cpp

@@ -184,7 +184,7 @@ roaring::Roaring64Map deserializeDeltaBitmapArray(std::string_view serializedPay
     return result;
   }
 
-  VELOX_USER_FAIL("Unexpected Delta bitmap array magic number {} for {}", magic, dvPath);
+  VELOX_USER_FAIL("Unexpected magic number {} for {}", magic, dvPath);
 }
 
 } // namespace

cpp/velox/compute/delta/DeltaSplit.h

@@ -97,7 +97,7 @@ struct DeltaDeletionVectorDescriptor {
   /// Format: If offset is None then <storageType><pathOrInlineDv>
   ///         Otherwise <storageType><pathOrInlineDv>@<offset>
   std::string uniqueId() const {
-    char storageTypeChar;
+    char storageTypeChar = 'u';
     switch (storageType) {
       case DeltaDeletionVectorStorageType::kUuidPath:
         storageTypeChar = 'u';

cpp/velox/compute/delta/DeltaUuidUtils.cpp

@@ -122,7 +122,7 @@ std::string DeltaUuidUtils::encodeBytesToBase85(std::string_view data) {
 }
 
 DeltaUuidUtils::Uuid DeltaUuidUtils::decodeZ85ToUuid(const std::string& z85) {
-  VELOX_CHECK_EQ(z85.length(), 20, "Z85 encoded UUID must be exactly 20 characters");
+  VELOX_USER_CHECK_EQ(z85.length(), 20, "Z85 encoded UUID must be exactly 20 characters");
 
   Uuid uuid;
 
@@ -136,11 +136,11 @@ DeltaUuidUtils::Uuid DeltaUuidUtils::decodeZ85ToUuid(const std::string& z85) {
 
 std::string DeltaUuidUtils::decodeBase85ToBytes(std::string_view encoded, size_t decodedSize) {
   if (encoded.empty()) {
-    VELOX_CHECK_EQ(decodedSize, 0, "Expected decoded Base85 size to be 0 for empty input");
+    VELOX_USER_CHECK_EQ(decodedSize, 0, "Expected decoded Base85 size to be 0 for empty input");
     return "";
   }
 
-  VELOX_CHECK_EQ(encoded.size() % 5, 0, "Base85 encoded payload must be aligned to 5-character blocks");
+  VELOX_USER_CHECK_EQ(encoded.size() % 5, 0, "Base85 encoded payload must be aligned to 5-character blocks");
 
   const size_t maxDecodedSize = (encoded.size() / 5) * 4;
   VELOX_CHECK_LE(
@@ -201,7 +201,7 @@ std::pair<std::string, DeltaUuidUtils::Uuid> DeltaUuidUtils::extractUuidFromZ85(
   // Z85-encoded UUID is always 20 characters
   // Any characters before that are the random prefix
   if (z85.length() < 20) {
-    VELOX_FAIL("Z85 string too short to contain UUID: {}", z85);
+    VELOX_USER_FAIL("Z85 string too short to contain UUID: {}", z85);
   }
 
   std::string randomPrefix;
@@ -223,13 +223,22 @@ std::pair<std::string, DeltaUuidUtils::Uuid> DeltaUuidUtils::extractUuidFromZ85(
 std::string
 DeltaUuidUtils::reconstructUuidPath(const std::string& tableDir, const std::string& randomPrefix, const Uuid& uuid) {
   std::ostringstream oss;
-  oss << tableDir;
+  const auto normalizedTableDir =
+      !tableDir.empty() && tableDir.back() == '/' ? tableDir.substr(0, tableDir.size() - 1) : tableDir;
+  oss << normalizedTableDir;
 
   if (!randomPrefix.empty()) {
-    oss << "/" << randomPrefix;
+    if (!normalizedTableDir.empty()) {
+      oss << "/";
+    }
+    oss << randomPrefix;
+  }
+
+  if (!normalizedTableDir.empty() || !randomPrefix.empty()) {
+    oss << "/";
   }
 
-  oss << "/deletion_vector_" << uuidToString(uuid) << ".bin";
+  oss << "deletion_vector_" << uuidToString(uuid) << ".bin";
   return oss.str();
 }
 

cpp/velox/compute/delta/tests/DeltaDeletionVectorReaderTest.cpp

@@ -573,7 +573,7 @@ TEST_F(DeltaDeletionVectorReaderTest, CardinalityValidationSuccess) {
   auto reader = std::make_unique<DeltaDeletionVectorReader>(nullptr, pool_.get(), ioStats_);
 
   // Should succeed with correct cardinality
-  EXPECT_NO_THROW(reader->loadInlineDeletionVector(inlineData, 5));
+  EXPECT_NO_THROW(reader->loadInlineDeletionVector(inlineData, std::nullopt, 5));
 
   // Verify cardinality
   EXPECT_EQ(reader->estimatedDeletedRowCount(), 5);
@@ -586,7 +586,7 @@ TEST_F(DeltaDeletionVectorReaderTest, CardinalityValidationMismatchThrows) {
   auto reader = std::make_unique<DeltaDeletionVectorReader>(nullptr, pool_.get(), ioStats_);
 
   // Should throw with incorrect cardinality
-  EXPECT_THROW(reader->loadInlineDeletionVector(inlineData, 3), VeloxUserError);
+  EXPECT_THROW(reader->loadInlineDeletionVector(inlineData, std::nullopt, 3), VeloxUserError);
 }
 
 TEST_F(DeltaDeletionVectorReaderTest, CardinalityValidationFileSuccess) {
@@ -635,7 +635,7 @@ TEST_F(DeltaDeletionVectorReaderTest, LargeCardinalityValidation) {
   auto reader = std::make_unique<DeltaDeletionVectorReader>(nullptr, pool_.get(), ioStats_);
 
   // Should succeed with correct cardinality
-  EXPECT_NO_THROW(reader->loadInlineDeletionVector(inlineData, 1000));
+  EXPECT_NO_THROW(reader->loadInlineDeletionVector(inlineData, std::nullopt, 1000));
   EXPECT_EQ(reader->estimatedDeletedRowCount(), 1000);
 }
 

cpp/velox/compute/delta/tests/DeltaUuidUtilsTest.cpp

@@ -61,7 +61,7 @@ TEST_F(DeltaUuidUtilsTest, DecodeZ85InvalidLength) {
 
 TEST_F(DeltaUuidUtilsTest, DecodeZ85InvalidCharacter) {
   // Test with invalid character (not in Z85 alphabet)
-  std::string encoded = "~~~~~"; // '~' is not in Z85 alphabet
+  std::string encoded(20, '~'); // '~' is not in Z85 alphabet
 
   EXPECT_THROW(DeltaUuidUtils::decodeZ85ToUuid(encoded), facebook::velox::VeloxUserError);
 }

gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala

@@ -19,10 +19,8 @@ package org.apache.gluten.extension
 import org.apache.gluten.execution.DeltaScanTransformer
 import org.apache.gluten.extension.columnar.FallbackTags
 import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
-import org.apache.gluten.sql.shims.SparkShimLoader
 
 import org.apache.spark.sql.delta.DeltaParquetFileFormat
-import org.apache.spark.sql.delta.DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME
 import org.apache.spark.sql.delta.files.TahoeFileIndex
 import org.apache.spark.sql.delta.stats.PreparedDeltaFileIndex
 import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
@@ -32,11 +30,8 @@ case class OffloadDeltaScan() extends OffloadSingleNode {
     case scan: FileSourceScanExec if isDeltaScan(scan) && isDeltaLogScan(scan) =>
       FallbackTags.add(scan, "fallback Delta _delta_log scan")
       scan
-    case scan: FileSourceScanExec if isDvPreparedDeltaScan(scan) =>
-      DeltaScanTransformer(scan)
     case scan: FileSourceScanExec if isDeltaScan(scan) =>
-      FallbackTags.add(scan, "fallback plain Delta scan without DV preprocessing")
-      scan
+      DeltaScanTransformer(scan)
     case other => other
   }
 
@@ -55,18 +50,6 @@ case class OffloadDeltaScan() extends OffloadSingleNode {
     scan.relation.location.isInstanceOf[PreparedDeltaFileIndex]
   }
 
-  private def isDvPreparedDeltaScan(scan: FileSourceScanExec): Boolean = {
-    isDeltaScan(scan) && hasDeletionVectorMarkers(scan)
-  }
-
-  private def hasDeletionVectorMarkers(scan: FileSourceScanExec): Boolean = {
-    val sparkShims = SparkShimLoader.getSparkShims
-    scan.output.exists(_.name == IS_ROW_DELETED_COLUMN_NAME) ||
-    scan.requiredSchema.fieldNames.contains(IS_ROW_DELETED_COLUMN_NAME) ||
-    scan.output.exists(attr => sparkShims.isRowIndexMetadataColumn(attr.name)) ||
-    scan.requiredSchema.fieldNames.exists(sparkShims.isRowIndexMetadataColumn)
-  }
-
   private def isDeltaLogScan(scan: FileSourceScanExec): Boolean = {
     scan.relation.location.rootPaths.exists {
       path =>