issue_ 8756_schemaevolution

import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.config._
import org.apache.hudi.common.model.HoodieRecord
import org.apache.hudi.keygen.constant._
import org.apache.hudi.sync.common._
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import org.apache.hudi.hive._;

val dbName = "default"
val table = "issue_8756_3"
val path = s"/tmp/${table}"

case class TestClass(bool_value: Boolean, timestamp_value: Timestamp)
Seq(
  ("1", "a", TestClass(true, Timestamp.from(LocalDateTime.now().toInstant(ZoneOffset.UTC)))),
  ("1", "b", TestClass(false, Timestamp.from(LocalDateTime.now().toInstant(ZoneOffset.UTC))))
).toDF("partition", "id", "detail")
  .write
  .format("org.apache.hudi")
  .options(
    Map(
      HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key -> "id",
      KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key -> "id",
      HoodieWriteConfig.TBL_NAME.key -> table,
      HiveSyncConfig.HIVE_SYNC_ENABLED.key -> "true",
      HiveSyncConfig.HIVE_URL.key -> "jdbc:hive2://hiveserver:10000",
      HiveSyncConfig.HIVE_AUTO_CREATE_DATABASE.key -> "true",
      HoodieSyncConfig.META_SYNC_DATABASE_NAME.key -> dbName,
      HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key -> classOf[NonPartitionedExtractor].getName,
      HoodieSyncConfig.META_SYNC_TABLE_NAME.key -> table,
      KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key -> "partition",
      KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key -> "true",
    )
  )
  .mode(SaveMode.Overwrite)
  .save(path)
  
spark.read.format("org.apache.hudi").load(path).show(false)

case class TestClass2(bool_value: String, timestamp_value: String)
Seq(
  ("2", "c", TestClass2("str1", LocalDateTime.now().toString)),
  ("2", "d", TestClass2("str2", LocalDateTime.now().toString))
).toDF("partition", "id", "detail")
  .write
  .format("org.apache.hudi")
  .options(
    Map(
      HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key -> "id",
      KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key -> "id",
      HoodieWriteConfig.TBL_NAME.key -> table,
      HiveSyncConfig.HIVE_SYNC_ENABLED.key -> "true",
      HiveSyncConfig.HIVE_URL.key -> "jdbc:hive2://hiveserver:10000",
      HiveSyncConfig.HIVE_AUTO_CREATE_DATABASE.key -> "true",
      HoodieSyncConfig.META_SYNC_DATABASE_NAME.key -> dbName,
      HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key -> classOf[NonPartitionedExtractor].getName,
      HoodieSyncConfig.META_SYNC_TABLE_NAME.key -> table,
      KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key -> "partition",
      KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key -> "true",
    )
  )
  .mode(SaveMode.Append)
  .save(path)

spark.read.format("org.apache.hudi").load(path).show(false)

==============ERROR IN WRITING===========================
Caused by: org.apache.hadoop.hive.ql.metadata.HiveException: Unable to alter table. The following columns have types incompatible with the existing columns in their respective positions :
detail
	at org.apache.hadoop.hive.ql.metadata.Hive.alterTable(Hive.java:627)
	at org.apache.hadoop.hive.ql.exec.DDLTask.alterTable(DDLTask.java:3590)
	at org.apache.hadoop.hive.ql.exec.DDLTask.execute(DDLTask.java:390)
	at org.apache.hadoop.hive.ql.exec.Task.executeTask(Task.java:199)
	at org.apache.hadoop.hive.ql.exec.TaskRunner.runSequential(TaskRunner.java:100)
	at org.apache.hadoop.hive.ql.Driver.launchTask(Driver.java:2183)
	at org.apache.hadoop.hive.ql.Driver.execute(Driver.java:1839)
	at org.apache.hadoop.hive.ql.Driver.runInternal(Driver.java:1526)
	at org.apache.hadoop.hive.ql.Driver.run(Driver.java:1237)
	at org.apache.hadoop.hive.ql.Driver.run(Driver.java:1232)
	at org.apache.hive.service.cli.operation.SQLOperation.runQuery(SQLOperation.java:255)
	... 11 more
=====================