diff --git a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala index 9c2b9efd5d6..2471a39161b 100644 --- a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala +++ b/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 diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 4e680fb25ba..842b7a567c6 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -525,6 +525,8 @@ void WholeStageResultIterator::collectMetrics() { std::unordered_map WholeStageResultIterator::getQueryContextConf() { std::unordered_map configs = {}; // Find batch size from Spark confs. If found, set the preferred and max batch size. + configs[velox::core::QueryConfig::kValidateOutputFromOperators] = + veloxCfg_->get(kValidateOutputFromOperators, kValidateOutputFromOperatorsDefault) ? "true" : "false"; configs[velox::core::QueryConfig::kPreferredOutputBatchRows] = std::to_string(veloxCfg_->get(kSparkBatchSize, 4096)); configs[velox::core::QueryConfig::kMaxOutputBatchRows] = diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index c0ae4b3ff86..bd97e22f2e4 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/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;