Incoming batch schema is not compatible with the table's one

I got below exception when ingest data from sql server into hudi.
org.apache.hudi.exception.SchemaCompatibilityException: Incoming batch schema is not compatible with the table's one
at org.apache.hudi.HoodieSparkSqlWriter$.deduceWriterSchema(HoodieSparkSqlWriter.scala:496)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:314)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:150)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:47)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:98)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:109)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:169)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:95)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:584)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:176)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:584)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)

Source table ddl is:
-- auto-generated definition
create table Address
(
Id int identity
constraint [xxxx]
primary key,
**Line1** nvarchar(128),
**Line2** nvarchar(128),
ccode nvarchar(2) not null
constraint [xxxx]
references Country,
XEID int not null,
cbUser nvarchar(48),
MuUser int not null,
MyUser nvarchar(48),
CreateDate datetime not null,
Latitude decimal(12, 9),
Longitude decimal(12, 9)
)

Environment Description

Hudi version : 0.9

Spark version : 3.0.1

Hive version : 3.1

Hadoop version : 3.2.2

Storage (HDFS/S3/GCS..) :

Running on Docker? no :

Start at HoodieSparkSqlWriter.scala lines 314 and 496, where the write path deduces the incoming schema and raises SchemaCompatibilityException. Compare the SQL Server source schema shown in the issue with the table schema and inspect existing schema-compatibility tests; done means the incompatibility is reproduced and its handling or documentation is covered by a focused test.

Written by the indexing model from the issue text.