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

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxInsertSuite.scala

@@ -0,0 +1,142 @@
+/*
+ * 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.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.config.GlutenConfig
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.internal.SQLConf
+
+class VeloxInsertSuite extends VeloxWholeStageTransformerSuite {
+  override protected val resourcePath: String = "placeholder"
+  override protected val fileFormat: String = "parquet"
+
+  override protected def sparkConf: SparkConf = {
+    super.sparkConf
+      .set("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+      .set("spark.sql.shuffle.partitions", "1")
+      .set("spark.memory.offHeap.size", "2g")
+      .set("spark.unsafe.exceptionOnMemoryLeak", "true")
+  }
+
+  test("storeAssignmentPolicy default ANSI is independent from ANSI mode") {
+    withTable("store_assignment_ansi_src", "store_assignment_ansi") {
+      withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+        assert(SQLConf.get.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.ANSI)
+
+        createTableWithValue("store_assignment_ansi_src", "STRING", "'2147483648'")
+        createTable("store_assignment_ansi", "INT")
+        assertUnsafeCastAnalysisException("STRING", "INT") {
+          insertIntoFrom("store_assignment_ansi", "store_assignment_ansi_src").collect()
+        }
+
+        withSQLConf(
+          SQLConf.STORE_ASSIGNMENT_POLICY.key -> SQLConf.StoreAssignmentPolicy.LEGACY.toString) {
+          val insert = insertIntoFrom("store_assignment_ansi", "store_assignment_ansi_src")
+          insert.collect()
+          checkGlutenPlan[ProjectExecTransformer](insert)
+          checkAnswer(spark.table("store_assignment_ansi"), Row(null))
+        }
+      }
+    }
+  }
+
+  test("storeAssignmentPolicy preserves configured cast modes") {
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+      withTable("store_assignment_ansi_src", "store_assignment_ansi") {
+        createTableWithValue("store_assignment_ansi_src", "STRING", "'2147483648'")
+        createTable("store_assignment_ansi", "INT")
+
+        withSQLConf(
+          SQLConf.STORE_ASSIGNMENT_POLICY.key -> SQLConf.StoreAssignmentPolicy.ANSI.toString) {
+          assertUnsafeCastAnalysisException("STRING", "INT") {
+            insertIntoFrom("store_assignment_ansi", "store_assignment_ansi_src").collect()
+          }
+          checkAnswer(spark.table("store_assignment_ansi"), Seq.empty[Row])
+        }
+      }
+    }
+
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+      withTable("store_assignment_legacy_src", "store_assignment_legacy") {
+        createTableWithValue("store_assignment_legacy_src", "STRING", "'2147483648'")
+        createTable("store_assignment_legacy", "INT")
+
+        // Disable the whole-plan ANSI fallback so the legacy store-assignment cast
+        // gets a chance to offload while the session runs in ANSI mode.
+        withSQLConf(
+          SQLConf.STORE_ASSIGNMENT_POLICY.key -> SQLConf.StoreAssignmentPolicy.LEGACY.toString,
+          GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false"
+        ) {
+          val insert = insertIntoFrom("store_assignment_legacy", "store_assignment_legacy_src")
+          insert.collect()
+          checkGlutenPlan[ProjectExecTransformer](insert)
+          checkAnswer(spark.table("store_assignment_legacy"), Row(null))
+        }
+      }
+    }
+  }
+
+  test("storeAssignmentPolicy strict rejects unsafe insert casts") {
+    withTable("store_assignment_strict_src", "store_assignment_strict") {
+      withSQLConf(
+        SQLConf.STORE_ASSIGNMENT_POLICY.key -> SQLConf.StoreAssignmentPolicy.STRICT.toString) {
+        createTableWithValue("store_assignment_strict_src", "INT", "1")
+        createTable("store_assignment_strict", "TINYINT")
+
+        assertUnsafeCastAnalysisException("INT", "TINYINT") {
+          insertIntoFrom("store_assignment_strict", "store_assignment_strict_src").collect()
+        }
+        checkAnswer(spark.table("store_assignment_strict"), Seq.empty[Row])
+      }
+    }
+  }
+
+  private def createTable(table: String, dataType: String): Unit =
+    spark.sql(s"CREATE TABLE $table (c $dataType) USING PARQUET")
+
+  private def createTableWithValue(table: String, dataType: String, value: String): Unit = {
+    createTable(table, dataType)
+    spark.sql(s"INSERT INTO $table VALUES ($value)").collect()
+  }
+
+  private def insertIntoFrom(target: String, source: String) =
+    spark.sql(s"INSERT INTO $target SELECT c FROM $source")
+
+  private def assertUnsafeCastAnalysisException(
+      fromType: String,
+      toType: String)(f: => Unit): Unit = {
+    val exception = intercept[AnalysisException](f)
+    // Spark 3.3/3.4 report the types in lower case, e.g. "string to int". Since Spark 3.5,
+    // the types in the exception message are quoted in upper case, e.g. "STRING" to "INT".
+    // The case conversion can be removed once Spark 3.4 and earlier are no longer supported.
+    val message = exceptionMessages(exception).toLowerCase()
+    assert(message.contains(fromType.toLowerCase()), message)
+    assert(message.contains(toType.toLowerCase()), message)
+    assert(message.contains("cast"), message)
+  }
+
+  private def exceptionMessages(e: Throwable): String = {
+    val message = Option(e.getMessage).getOrElse("")
+    if (e.getCause == null) {
+      message
+    } else {
+      message + "\n" + exceptionMessages(e.getCause)
+    }
+  }
+}

cpp/velox/operators/functions/RegistrationAllFunctions.cc

@@ -33,6 +33,7 @@
 #include "velox/functions/sparksql/Rand.h"
 #include "velox/functions/sparksql/aggregates/Register.h"
 #include "velox/functions/sparksql/registration/Register.h"
+#include "velox/functions/sparksql/specialforms/SparkCastExpr.h"
 #include "velox/functions/sparksql/window/WindowFunctionsRegistration.h"
 
 using namespace facebook;
@@ -84,6 +85,7 @@ void registerFunctionOverwrite() {
 
 void registerAllFunctions() {
   velox::functions::sparksql::registerFunctions("");
+  velox::functions::sparksql::registerSparkCastModeSpecialForms();
   velox::aggregate::prestosql::registerAllAggregateFunctions(
       "", true /*registerCompanionFunctions*/, false /*onlyPrestoSignatures*/, true /*overwrite*/);
   velox::functions::aggregate::sparksql::registerAggregateFunctions(

cpp/velox/substrait/SubstraitToVeloxExpr.cc

@@ -17,13 +17,17 @@
 
 #include "SubstraitToVeloxExpr.h"
 #include "TypeUtils.h"
+#include "velox/functions/sparksql/specialforms/SparkCastExpr.h"
 #include "velox/type/Timestamp.h"
 #include "velox/vector/FlatVector.h"
 #include "velox/vector/VariantToVector.h"
 
 using namespace facebook::velox;
 
 namespace {
+constexpr const char* kSparkAnsiCast = "spark_ansi_cast";
+constexpr const char* kSparkLegacyCast = "spark_legacy_cast";
+
 ArrayVectorPtr makeArrayVector(const VectorPtr& elements) {
   BufferPtr offsets = allocateOffsets(1, elements->pool());
   BufferPtr sizes = allocateOffsets(1, elements->pool());
@@ -147,14 +151,20 @@ 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));
   }
@@ -581,7 +591,20 @@ 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: {
+      const auto castName = functions::sparksql::SparkCastCallToSpecialForm::isAnsiSupported(inputs[0]->type(), type)
+          ? kSparkAnsiCast
+          : kSparkLegacyCast;
+      return std::make_shared<const core::CallTypedExpr>(type, std::move(inputs), castName);
+    }
+    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

@@ -15,14 +15,27 @@
  * limitations under the License.
  */
 
+#include <string>
 #include <vector>
 
 #include "operators/functions/RegistrationAllFunctions.h"
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/core/Expressions.h"
+#include "velox/functions/sparksql/SparkQueryConfig.h"
 #include "velox/functions/sparksql/tests/SparkFunctionBaseTest.h"
 
 using namespace facebook::velox::functions::sparksql::test;
 using namespace facebook::velox;
 
+namespace {
+constexpr const char* kSparkAnsiCast = "spark_ansi_cast";
+constexpr const char* kSparkLegacyCast = "spark_legacy_cast";
+
+std::string sparkAnsiEnabledConfigKey() {
+  return functions::sparksql::SparkQueryConfig::qualify(functions::sparksql::SparkQueryConfig::kAnsiEnabled);
+}
+} // namespace
+
 class SparkFunctionTest : public SparkFunctionBaseTest {
  public:
   SparkFunctionTest() {
@@ -111,3 +124,23 @@ TEST_F(SparkFunctionTest, roundWithDecimal) {
   runRoundWithDecimalTest<int16_t>(testRoundWithDecIntegralData<int16_t>());
   runRoundWithDecimalTest<int8_t>(testRoundWithDecIntegralData<int8_t>());
 }
+
+TEST_F(SparkFunctionTest, expressionLevelAnsiCastIgnoresSessionAnsiOff) {
+  queryCtx_->testingOverrideConfigUnsafe({{sparkAnsiEnabledConfigKey(), "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}, kSparkAnsiCast);
+
+  VELOX_ASSERT_THROW(evaluate(ansiCast, input), "Cannot cast");
+}
+
+TEST_F(SparkFunctionTest, expressionLevelLegacyCastIgnoresSessionAnsiOn) {
+  queryCtx_->testingOverrideConfigUnsafe({{sparkAnsiEnabledConfigKey(), "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}, kSparkLegacyCast);
+
+  facebook::velox::test::assertEqualVectors(makeFlatVector<int8_t>({-121}), evaluate(legacyCast, input));
+}

cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc

@@ -269,7 +269,7 @@ TEST_F(Substrait2VeloxPlanConversionTest, ifthenTest) {
   // Convert to Velox PlanNode.
   auto planNode = planConverter_->toVeloxPlan(substraitPlan, std::vector<::substrait::ReadRel_LocalFiles>{split});
   ASSERT_EQ(
-      "-- Project[1][expressions: ] -> \n  -- TableScan[0][table: hive_table, remaining filter: (and(and(and(and(isnotnull(\"hd_vehicle_count\"),or(equalto(\"hd_buy_potential\",>10000),equalto(\"hd_buy_potential\",unknown))),greaterthan(\"hd_vehicle_count\",0)),if(greaterthan(\"hd_vehicle_count\",0),greaterthan(divide(cast(\"hd_dep_count\" as DOUBLE),cast(\"hd_vehicle_count\" as DOUBLE)),1.2))),isnotnull(\"hd_demo_sk\"))), data columns: ROW<hd_demo_sk:BIGINT,hd_buy_potential:VARCHAR,hd_dep_count:BIGINT,hd_vehicle_count:BIGINT>] -> n0_0:BIGINT, n0_1:VARCHAR, n0_2:BIGINT, n0_3:BIGINT\n",
+      "-- Project[1][expressions: ] -> \n  -- TableScan[0][table: hive_table, remaining filter: (and(and(and(and(isnotnull(\"hd_vehicle_count\"),or(equalto(\"hd_buy_potential\",>10000),equalto(\"hd_buy_potential\",unknown))),greaterthan(\"hd_vehicle_count\",0)),if(greaterthan(\"hd_vehicle_count\",0),greaterthan(divide(spark_legacy_cast(\"hd_dep_count\"),spark_legacy_cast(\"hd_vehicle_count\")),1.2))),isnotnull(\"hd_demo_sk\"))), data columns: ROW<hd_demo_sk:BIGINT,hd_buy_potential:VARCHAR,hd_dep_count:BIGINT,hd_vehicle_count:BIGINT>] -> n0_0:BIGINT, n0_1:VARCHAR, n0_2:BIGINT, n0_3:BIGINT\n",
       planNode->toString(true, true));
 }
 

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)
   }
 }
 

shims/spark33/src/main/scala/org/apache/gluten/sql/shims/spark33/Spark33Shims.scala

@@ -210,6 +210,15 @@ class Spark33Shims extends SparkShims {
     }
   }
 
+  override def withAnsiEvalMode(expr: Expression): Boolean = {
+    expr match {
+      // Use the cast's own flag rather than the session conf: store-assignment casts
+      // can carry ansiEnabled = false even when the session runs in ANSI mode.
+      case c: Cast => c.ansiEnabled
+      case _ => false
+    }
+  }
+
   override def createParquetFilters(
       conf: SQLConf,
       schema: MessageType,