[GLUTEN-10134][VL] Preserve store assignment cast modes (STORE_ASSIGNMENT_POLICY defaults to ANSI)

cpp/velox/CMakeLists.txt

@@ -170,6 +170,7 @@ set(VELOX_SRCS
     memory/VeloxMemoryManager.cc
     operators/functions/RegistrationAllFunctions.cc
     operators/functions/RowConstructorWithNull.cc
+    operators/functions/SparkCastModeSpecialForms.cc
     operators/functions/SparkExprToSubfieldFilterParser.cc
     operators/plannodes/RowVectorStream.cc
     operators/hashjoin/HashTableBuilder.cc

cpp/velox/operators/functions/RegistrationAllFunctions.cc

@@ -19,6 +19,7 @@
 #include "operators/functions/Arithmetic.h"
 #include "operators/functions/RowConstructorWithNull.h"
 #include "operators/functions/RowFunctionWithNull.h"
+#include "operators/functions/SparkCastModeSpecialForms.h"
 #include "velox/expression/SpecialFormRegistry.h"
 #include "velox/expression/VectorFunction.h"
 #include "velox/functions/iceberg/Register.h"
@@ -83,6 +84,7 @@ void registerFunctionOverwrite() {
 
 void registerAllFunctions() {
   velox::functions::sparksql::registerFunctions("");
+  registerSparkCastModeSpecialForms();
   velox::aggregate::prestosql::registerAllAggregateFunctions(
       "", true /*registerCompanionFunctions*/, false /*onlyPrestoSignatures*/, true /*overwrite*/);
   velox::functions::aggregate::sparksql::registerAggregateFunctions(

cpp/velox/operators/functions/SparkCastModeSpecialForms.cc

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "operators/functions/SparkCastModeSpecialForms.h"
+
+#include "velox/expression/SpecialFormRegistry.h"
+#include "velox/functions/sparksql/specialforms/SparkCastExpr.h"
+#include "velox/functions/sparksql/specialforms/SparkCastHooks.h"
+
+namespace gluten {
+namespace {
+
+using namespace facebook::velox;
+using facebook::velox::functions::sparksql::SparkCastExpr;
+using facebook::velox::functions::sparksql::SparkCastHooks;
+
+bool isIntegralType(const TypePtr& type) {
+  return type == TINYINT() || type == SMALLINT() || type == INTEGER() ||
+      type == BIGINT();
+}
+
+bool isAnsiSupported(const TypePtr& fromType, const TypePtr& toType) {
+  if (fromType->isVarchar()) {
+    return toType->isBoolean() || toType->isDate() || isIntegralType(toType);
+  }
+  return false;
+}
+
+exec::ExprPtr makeSparkCastExpr(
+    const TypePtr& type,
+    exec::ExprPtr&& input,
+    bool trackCpuUsage,
+    bool isTryCast,
+    bool allowOverflow,
+    const core::QueryConfig& config) {
+  return std::make_shared<SparkCastExpr>(
+      type,
+      std::move(input),
+      trackCpuUsage,
+      isTryCast,
+      std::make_shared<SparkCastHooks>(config, allowOverflow));
+}
+
+class SparkAnsiCastCallToSpecialForm : public exec::CastCallToSpecialForm {
+ public:
+  exec::ExprPtr constructSpecialForm(
+      const TypePtr& type,
+      std::vector<exec::ExprPtr>&& compiledChildren,
+      bool trackCpuUsage,
+      const core::QueryConfig& config) override {
+    VELOX_CHECK_EQ(
+        compiledChildren.size(),
+        1,
+        "ANSI CAST statements expect exactly 1 argument, received {}.",
+        compiledChildren.size());
+
+    const auto& fromType = compiledChildren[0]->type();
+    const bool isTryCast = !isAnsiSupported(fromType, type);
+    return makeSparkCastExpr(
+        type,
+        std::move(compiledChildren[0]),
+        trackCpuUsage,
+        isTryCast,
+        isTryCast,
+        config);
+  }
+};
+
+class SparkLegacyCastCallToSpecialForm : public exec::CastCallToSpecialForm {
+ public:
+  exec::ExprPtr constructSpecialForm(
+      const TypePtr& type,
+      std::vector<exec::ExprPtr>&& compiledChildren,
+      bool trackCpuUsage,
+      const core::QueryConfig& config) override {
+    VELOX_CHECK_EQ(
+        compiledChildren.size(),
+        1,
+        "LEGACY CAST statements expect exactly 1 argument, received {}.",
+        compiledChildren.size());
+
+    return makeSparkCastExpr(
+        type,
+        std::move(compiledChildren[0]),
+        trackCpuUsage,
+        true,
+        true,
+        config);
+  }
+};
+
+} // namespace
+
+void registerSparkCastModeSpecialForms() {
+  exec::registerFunctionCallToSpecialForm(
+      kSparkAnsiCast, std::make_unique<SparkAnsiCastCallToSpecialForm>());
+  exec::registerFunctionCallToSpecialForm(
+      kSparkLegacyCast, std::make_unique<SparkLegacyCastCallToSpecialForm>());
+}
+
+} // namespace gluten

cpp/velox/operators/functions/SparkCastModeSpecialForms.h

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+namespace gluten {
+
+constexpr const char* kSparkAnsiCast = "spark_ansi_cast";
+constexpr const char* kSparkLegacyCast = "spark_legacy_cast";
+
+void registerSparkCastModeSpecialForms();
+
+} // namespace gluten

cpp/velox/substrait/SubstraitToVeloxExpr.cc

@@ -17,6 +17,7 @@
 
 #include "SubstraitToVeloxExpr.h"
 #include "TypeUtils.h"
+#include "operators/functions/SparkCastModeSpecialForms.h"
 #include "velox/vector/FlatVector.h"
 #include "velox/vector/VariantToVector.h"
 
@@ -146,16 +147,24 @@ TypePtr getScalarType(const ::substrait::Expression::Literal& literal) {
   }
 }
 
-/// Whether is try cast.
-bool isTryCast(::substrait::Expression::Cast::FailureBehavior failureBehavior) {
+enum class SparkCastMode {
+  kLegacy,
+  kAnsi,
+  kTry,
+};
+
+SparkCastMode sparkCastMode(
+    ::substrait::Expression::Cast::FailureBehavior failureBehavior) {
   switch (failureBehavior) {
     case ::substrait::Expression_Cast_FailureBehavior_FAILURE_BEHAVIOR_UNSPECIFIED:
+      return SparkCastMode::kLegacy;
     case ::substrait::Expression_Cast_FailureBehavior_FAILURE_BEHAVIOR_THROW_EXCEPTION:
-      return false;
+      return SparkCastMode::kAnsi;
     case ::substrait::Expression_Cast_FailureBehavior_FAILURE_BEHAVIOR_RETURN_NULL:
-      return true;
+      return SparkCastMode::kTry;
     default:
-      VELOX_NYI("The given failure behavior is NOT supported: '{}'", std::to_string(failureBehavior));
+      VELOX_NYI(
+          "The given failure behavior is NOT supported: '{}'", std::to_string(failureBehavior));
   }
 }
 
@@ -564,7 +573,18 @@ core::TypedExprPtr SubstraitVeloxExprConverter::toVeloxExpr(
     const RowTypePtr& inputType) {
   auto type = SubstraitParser::parseType(castExpr.type());
   std::vector<core::TypedExprPtr> inputs{toVeloxExpr(castExpr.input(), inputType)};
-  return std::make_shared<core::CastTypedExpr>(type, inputs, isTryCast(castExpr.failure_behavior()));
+  switch (sparkCastMode(castExpr.failure_behavior())) {
+    case SparkCastMode::kLegacy:
+      return std::make_shared<const core::CallTypedExpr>(
+          type, std::move(inputs), kSparkLegacyCast);
+    case SparkCastMode::kAnsi:
+      return std::make_shared<const core::CallTypedExpr>(
+          type, std::move(inputs), kSparkAnsiCast);
+    case SparkCastMode::kTry:
+      return std::make_shared<core::CastTypedExpr>(type, std::move(inputs), true);
+    default:
+      VELOX_UNREACHABLE();
+  }
 }
 
 core::TypedExprPtr SubstraitVeloxExprConverter::toVeloxExpr(

cpp/velox/tests/SparkFunctionTest.cc

@@ -18,6 +18,10 @@
 #include <vector>
 
 #include "operators/functions/RegistrationAllFunctions.h"
+#include "operators/functions/SparkCastModeSpecialForms.h"
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/core/Expressions.h"
+#include "velox/core/QueryConfig.h"
 #include "velox/functions/sparksql/tests/SparkFunctionBaseTest.h"
 
 using namespace facebook::velox::functions::sparksql::test;
@@ -111,3 +115,32 @@ TEST_F(SparkFunctionTest, roundWithDecimal) {
   runRoundWithDecimalTest<int16_t>(testRoundWithDecIntegralData<int16_t>());
   runRoundWithDecimalTest<int8_t>(testRoundWithDecIntegralData<int8_t>());
 }
+
+TEST_F(SparkFunctionTest, expressionLevelAnsiCastIgnoresSessionAnsiOff) {
+  queryCtx_->testingOverrideConfigUnsafe(
+      {{core::QueryConfig::kSparkAnsiEnabled, "false"}});
+  auto input = makeRowVector({makeFlatVector<std::string>({"2147483648"})});
+  core::TypedExprPtr field =
+      std::make_shared<const core::FieldAccessTypedExpr>(VARCHAR(), "c0");
+  auto ansiCast = std::make_shared<const core::CallTypedExpr>(
+      INTEGER(),
+      std::vector<core::TypedExprPtr>{field},
+      gluten::kSparkAnsiCast);
+
+  VELOX_ASSERT_THROW(evaluate(ansiCast, input), "Cannot cast");
+}
+
+TEST_F(SparkFunctionTest, expressionLevelLegacyCastIgnoresSessionAnsiOn) {
+  queryCtx_->testingOverrideConfigUnsafe(
+      {{core::QueryConfig::kSparkAnsiEnabled, "true"}});
+  auto input = makeRowVector({makeFlatVector<int32_t>({1234567})});
+  core::TypedExprPtr field =
+      std::make_shared<const core::FieldAccessTypedExpr>(INTEGER(), "c0");
+  auto legacyCast = std::make_shared<const core::CallTypedExpr>(
+      TINYINT(),
+      std::vector<core::TypedExprPtr>{field},
+      gluten::kSparkLegacyCast);
+
+  facebook::velox::test::assertEqualVectors(
+      makeFlatVector<int8_t>({-121}), evaluate(legacyCast, input));
+}

gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/CastNode.java

@@ -23,28 +23,46 @@
 import java.io.Serializable;
 
 public class CastNode implements ExpressionNode, Serializable {
+  public enum CastMode {
+    LEGACY,
+    ANSI,
+    TRY
+  }
+
   private final TypeNode typeNode;
   private final ExpressionNode expressionNode;
 
-  public final boolean isTryCast;
+  public final CastMode castMode;
 
   CastNode(TypeNode typeNode, ExpressionNode expressionNode, boolean isTryCast) {
+    this(typeNode, expressionNode, isTryCast ? CastMode.TRY : CastMode.ANSI);
+  }
+
+  CastNode(TypeNode typeNode, ExpressionNode expressionNode, CastMode castMode) {
     this.typeNode = typeNode;
     this.expressionNode = expressionNode;
-    this.isTryCast = isTryCast;
+    this.castMode = castMode;
   }
 
   @Override
   public Expression toProtobuf() {
     Expression.Cast.Builder castBuilder = Expression.Cast.newBuilder();
     castBuilder.setType(typeNode.toProtobuf());
     castBuilder.setInput(expressionNode.toProtobuf());
-    if (!isTryCast) {
-      // Throw exception on failure.
-      castBuilder.setFailureBehaviorValue(2);
-    } else {
-      // Return null on failure.
-      castBuilder.setFailureBehaviorValue(1);
+    switch (castMode) {
+      case ANSI:
+        // Throw exception on failure.
+        castBuilder.setFailureBehaviorValue(2);
+        break;
+      case TRY:
+        // Return null on failure.
+        castBuilder.setFailureBehaviorValue(1);
+        break;
+      case LEGACY:
+        // Leave failure behavior unspecified to preserve Spark legacy cast semantics.
+        break;
+      default:
+        throw new IllegalStateException("Unsupported cast mode: " + castMode);
     }
     Expression.Builder builder = Expression.newBuilder();
     builder.setCast(castBuilder.build());

gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/ExpressionBuilder.java

@@ -242,6 +242,11 @@ public static CastNode makeCast(
     return new CastNode(typeNode, expressionNode, isTryCast);
   }
 
+  public static CastNode makeCast(
+      TypeNode typeNode, ExpressionNode expressionNode, CastNode.CastMode castMode) {
+    return new CastNode(typeNode, expressionNode, castMode);
+  }
+
   public static StringMapNode makeStringMap(Map<String, String> values) {
     return new StringMapNode(values);
   }

gluten-substrait/src/main/scala/org/apache/gluten/expression/UnaryExpressionTransformer.scala

@@ -22,7 +22,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.gluten.substrait.`type`.ListNode
 import org.apache.gluten.substrait.`type`.MapNode
 import org.apache.gluten.substrait.SubstraitContext
-import org.apache.gluten.substrait.expression.{ExpressionBuilder, ExpressionNode, StructLiteralNode}
+import org.apache.gluten.substrait.expression.{CastNode, ExpressionBuilder, ExpressionNode, StructLiteralNode}
 
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.types._
@@ -45,10 +45,19 @@ case class CastTransformer(substraitExprName: String, child: ExpressionTransform
   extends UnaryExpressionTransformer {
   override def doTransform(context: SubstraitContext): ExpressionNode = {
     val typeNode = ConverterUtils.getTypeNode(dataType, original.nullable)
+    val sparkShims = SparkShimLoader.getSparkShims
+    // Store-assignment casts can carry EvalMode.ANSI even when session ANSI is disabled.
+    val castMode = if (sparkShims.withTryEvalMode(original)) {
+      CastNode.CastMode.TRY
+    } else if (sparkShims.withAnsiEvalMode(original)) {
+      CastNode.CastMode.ANSI
+    } else {
+      CastNode.CastMode.LEGACY
+    }
     ExpressionBuilder.makeCast(
       typeNode,
       child.doTransform(context),
-      SparkShimLoader.getSparkShims.withTryEvalMode(original))
+      castMode)
   }
 }
 

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/sources/GlutenInsertSuite.scala

@@ -124,6 +124,25 @@ class GlutenInsertSuite
     }
   }
 
+  testGluten("storeAssignmentPolicy default ANSI is independent from ANSI mode") {
+    withTable("store_assignment_ansi") {
+      withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+        assert(SQLConf.get.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.ANSI)
+
+        spark.sql("CREATE TABLE store_assignment_ansi (c INT) USING PARQUET")
+        intercept[Exception] {
+          spark.sql("INSERT INTO store_assignment_ansi SELECT '2147483648'").collect()
+        }
+
+        withSQLConf(
+          SQLConf.STORE_ASSIGNMENT_POLICY.key -> SQLConf.StoreAssignmentPolicy.LEGACY.toString) {
+          spark.sql("INSERT INTO store_assignment_ansi SELECT '2147483648'").collect()
+          checkAnswer(spark.table("store_assignment_ansi"), Row(null))
+        }
+      }
+    }
+  }
+
   ignoreGluten("Cleanup staging files if job failed") {
     // Using a unique table name in this test. Sometimes, the table is not removed for some unknown
     // reason, which can cause test failure (location already exists) if other following tests have