hive cli execute sql query hadoopcatalog table occurred timestamp type cast exception

[kukuzidian]

1. use spark stream api create hadoopcatalog tableA . tableA has 5 culumns ，one is TimestampType.withoutZone .
   I set this value create by TimestampData.fromEpochMillis(System.currentTimeMills()) .
2. I use hive command line client , test sql 'select * from tableA' ,this sql execute is OK. but test 'select count(*) from tableA' occurred exception . eg : java.lang.ClassCastException: java.time.LocalDateTime cannot be cast to java.sql.Timestamp

hive-2.3.7
iceberg-0.10.0
spark-3.0.1

[zhangdove]

The environment:

hive-2.3.7
iceberg-0.10.0
spark-3.0.0

1. Spark Conf

    val spark = SparkSession
      .builder()
      .master("local[2]")
      .appName("IcebergAPI")
      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
      .config("spark.sql.catalog.hadoop_prod.warehouse", "file:///Users/dovezhang/iceberg/warehouse")
      .getOrCreate()
    val conf: Configuration = new Configuration()
    val catalog: HadoopCatalog = new HadoopCatalog(conf, "file:///Users/dovezhang/iceberg/warehouse")
    val nameSpace = Namespace.of(schemaName)
    val tableIdentifier: TableIdentifier = TableIdentifier.of(nameSpace, tableName)

2. Spark Create Table:

  def createPartitionTable(catalog: HadoopCatalog, tableIdentifier: TableIdentifier): Unit = {
    val columns: List[Types.NestedField] = new ArrayList[Types.NestedField]
    columns.add(Types.NestedField.of(1, true, "id", Types.IntegerType.get, "id doc"))
    columns.add(Types.NestedField.of(2, true, "name", Types.StringType.get, "name doc"))
    columns.add(Types.NestedField.of(3, true, "time", Types.TimestampType.withZone(), "create time doc"))

    val schema: Schema = new Schema(columns)
    val partition = PartitionSpec.unpartitioned()

    val table = catalog.createTable(tableIdentifier, schema, partition)
  }

3. Spark write data

  case class DbTb(id: Int, name: String, time: Timestamp)

  def writeDataToIcebergHdfs(spark: SparkSession): Unit = {
    val seq = Seq(DbTb(1, "doveHDFS", Timestamp.valueOf("2020-07-06 13:40:00")),
      DbTb(2, "IcebergNameDoveHDFS", Timestamp.valueOf("2020-07-06 14:30:00")),
      DbTb(3, "SparkHDFS", Timestamp.valueOf("2020-07-06 15:20:00")))
    val structedTbDf = spark.createDataFrame(seq).toDF("id", "name", "time")

    import org.apache.spark.sql.functions
    structedTbDf.writeTo(s"hadoop_prod.${schemaName}.${tableName}").overwrite(functions.lit(true))
  }

4. Hive Client Query:

0: jdbc:hive2://localhost:10000> add jar /Users/dovezhang/software/idea/github/iceberg/hive-runtime/build/libs/iceberg-hive-runtime-0.10.0.jar;
0: jdbc:hive2://localhost:10000> create database testDb;
0: jdbc:hive2://localhost:10000> use testDb;
0: jdbc:hive2://localhost:10000> CREATE EXTERNAL TABLE testTb
. . . . . . . . . . . . . . . .> STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
. . . . . . . . . . . . . . . .> LOCATION 'file:///Users/dovezhang/iceberg/warehouse/testDb/testTb';
0: jdbc:hive2://localhost:10000> select * from testDb.testTb;
+------------+----------------------+------------------------+
| testtb.id  |     testtb.name      |      testtb.time       |
+------------+----------------------+------------------------+
| 1          | doveHDFS             | 2020-07-06 05:40:00.0  |
| 2          | IcebergNameDoveHDFS  | 2020-07-06 06:30:00.0  |
| 3          | SparkHDFS            | 2020-07-06 07:20:00.0  |
+------------+----------------------+------------------------+
0: jdbc:hive2://localhost:10000> select count(*) from testDb.testTb;
+------+
| _c0  |
+------+
| 3    |
+------+
1 row selected (27.854 seconds)

I have done a test here, and there is no problem. Is your TimestampData customized? Perhaps you can provide more detailed operation?

Good Luck!

[kukuzidian]

1. in iceberg 0.10.0 branch IcebergTimestampObjectInspector.java line 75 :
   public Object copyObject(Object o) {
   if (o == null) {
   return null;
   }

   Timestamp ts = (Timestamp) o;
   Timestamp copy = new Timestamp(ts.getTime());
   copy.setNanos(ts.getNanos());
   return copy;
   }

2. but in iceberg master branch IcebergTimestampObjectInspector.java line 75 :
   public Object copyObject(Object o) {
   if (o == null) {
   return null;
   }

   if (o instanceof Timestamp) {
   Timestamp ts = (Timestamp) o;
   Timestamp copy = new Timestamp(ts.getTime());
   copy.setNanos(ts.getNanos());
   return copy;
   } else if (o instanceof OffsetDateTime) {
   return OffsetDateTime.of(((OffsetDateTime) o).toLocalDateTime(), ((OffsetDateTime) o).getOffset());
   } else if (o instanceof LocalDateTime) {
   return LocalDateTime.of(((LocalDateTime) o).toLocalDate(), ((LocalDateTime) o).toLocalTime());
   } else {
   return o;
   }
   }

so when i get this exception , I find code in master brach is different from 0.10.0 branch. Then i use new code in master branch can read parquest file ok . so I guess this case already finshed in master .

[kukuzidian]

because in my company the network can not connect to the internet . so I can't copy the full exception message. and this issues can't upload pictures . I find it difficult to new a issues. @zhangdove

[pvary]

@kukuzidian: I think #1740 created by @lcspinter fixes the issue. It did not make it into 0.10.0 😢, so you have to build iceberg yourself until we have another release.

[kukuzidian]

@pvary @zhangdove OK , Thanks , now I close this issues .