[GLUTEN-12377][VL][DO NOT MERGE] Run the Delta suite with Velox operator output validation

backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala

@@ -55,6 +55,19 @@ trait DeltaSQLCommandTest extends SharedSparkSession {
       .set(VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key, "true")
       .set("spark.databricks.delta.snapshotPartitions", "2")
       .set("spark.gluten.sql.fallbackUnexpectedMetadataParquet", "true")
+      // Validate every operator's output vector. A Delta deletion-vector write scans only
+      // synthesized columns with a pushed-down filter, and on that path Velox emits a
+      // RowVector whose row-index child has no rows; the child is then wrapped in a
+      // dictionary, and reading it goes out of bounds and yields arbitrary heap values.
+      // Without this the failure is intermittent and reports a meaningless row index, which
+      // is how it went unnoticed for so long. With it, the scan is caught the moment it
+      // produces the malformed vector, so the suite fails deterministically and names the
+      // operator responsible.
+      //   Velox issue: https://github.com/facebookincubator/velox/issues/18535
+      //   Velox fix:   https://github.com/facebookincubator/velox/pull/18536
+      // Remove this once that fix is picked up, at which point the suite must go green
+      // again -- which is what validates the fix.
+      .set("spark.gluten.sql.columnar.backend.velox.validateOutputFromOperators", "true")
   }
 }
 // spotless:on

cpp/velox/compute/WholeStageResultIterator.cc

@@ -525,6 +525,8 @@ void WholeStageResultIterator::collectMetrics() {
 std::unordered_map<std::string, std::string> WholeStageResultIterator::getQueryContextConf() {
   std::unordered_map<std::string, std::string> configs = {};
   // Find batch size from Spark confs. If found, set the preferred and max batch size.
+  configs[velox::core::QueryConfig::kValidateOutputFromOperators] =
+      veloxCfg_->get<bool>(kValidateOutputFromOperators, kValidateOutputFromOperatorsDefault) ? "true" : "false";
   configs[velox::core::QueryConfig::kPreferredOutputBatchRows] =
       std::to_string(veloxCfg_->get<uint32_t>(kSparkBatchSize, 4096));
   configs[velox::core::QueryConfig::kMaxOutputBatchRows] =

cpp/velox/config/VeloxConfig.h

@@ -107,6 +107,14 @@ const std::string kValueStreamDynamicFilterEnabled =
     "spark.gluten.sql.columnar.backend.velox.valueStream.dynamicFilter.enabled";
 const bool kValueStreamDynamicFilterEnabledDefault = false;
 
+/// Turns on Velox's per-operator output vector validation
+/// (`debug.validate_output_from_operators`). Every operator's output is checked for
+/// structural consistency -- among other things that a dictionary's indexes address rows its
+/// base vector actually has -- and the first operator to emit a malformed vector is named.
+/// Debugging aid only: it costs a pass over every output batch, so it is off by default.
+const std::string kValidateOutputFromOperators = "spark.gluten.sql.columnar.backend.velox.validateOutputFromOperators";
+const bool kValidateOutputFromOperatorsDefault = false;
+
 const std::string kShowTaskMetricsWhenFinished = "spark.gluten.sql.columnar.backend.velox.showTaskMetricsWhenFinished";
 const bool kShowTaskMetricsWhenFinishedDefault = false;