[ https://issues.apache.org/jira/browse/FLINK-15363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17011343#comment-17011343 ]
Zhenghua Gao commented on FLINK-15363: -------------------------------------- Currently *HBaseTableSchema* use TypeInformation to specify an HBase table's schema, which would cause precision/scale loss for several data types. Meanwhile *HBaseTypeUtils* serialize a java.sql.Timestamp to long in bytes, which would cause precision loss for TIMESTAMP types. the hbase connector should use new type system to fix this. > Hbase connector do not support datatypes with precision like TIMESTAMP(9) and > DECIMAL(10,4) > ------------------------------------------------------------------------------------------- > > Key: FLINK-15363 > URL: https://issues.apache.org/jira/browse/FLINK-15363 > Project: Flink > Issue Type: Bug > Components: Connectors / HBase > Affects Versions: 1.10.0 > Reporter: Leonard Xu > Priority: Major > Fix For: 1.10.0 > > > {code:java} > // exception msg > rowtype of new rel:rowtype of new rel:RecordType(VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" order_id, VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" item, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, > DECIMAL(10, 4) amount, TIMESTAMP(3) order_time, TIME ATTRIBUTE(PROCTIME) NOT > NULL proc_time, DECIMAL(20, 4) amount_kg, TIME ATTRIBUTE(ROWTIME) ts, BIGINT > currency_id, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency_name, > DECIMAL(38, 4) rate, TIMESTAMP(3) currency_time, VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" country, VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" rowkey, RecordType:peek_no_expand(INTEGER country_id, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" country_name, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" country_name_cn, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" region_name) f1, > RecordType:peek_no_expand(TIMESTAMP(3) record_timestamp3, TIMESTAMP(3) > record_timestamp9, TIME(0) time3, TIME(0) time9, DECIMAL(38, 18) gdp) f2) NOT > NULLrowtype of set:RecordType(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > order_id, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" item, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, DECIMAL(10, 4) amount, > TIMESTAMP(3) order_time, TIME ATTRIBUTE(PROCTIME) NOT NULL proc_time, > DECIMAL(20, 4) amount_kg, TIME ATTRIBUTE(ROWTIME) ts, BIGINT currency_id, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency_name, DECIMAL(38, 4) > rate, TIMESTAMP(3) currency_time, VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" country, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" rowkey, > RecordType:peek_no_expand(INTEGER country_id, VARCHAR(2147483647) CHARACTER > SET "UTF-16LE" country_name, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > country_name_cn, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" region_name) f1, > RecordType:peek_no_expand(TIMESTAMP(3) record_timestamp3, TIMESTAMP(9) > record_timestamp9, TIME(0) time3, TIME(0) time9, DECIMAL(10, 4) gdp) f2) NOT > NULL at > org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:84) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:62) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at scala.collection.Iterator$class.foreach(Iterator.scala:893) at > scala.collection.AbstractIterator.foreach(Iterator.scala:1336) at > scala.collection.IterableLike$class.foreach(IterableLike.scala:72) at > scala.collection.AbstractIterable.foreach(Iterable.scala:54) at > scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157) at > scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104) at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:57) > at > org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:167) > at > org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:89) > at > org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77) > at > org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:223) > at > org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:150) > at > org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.java:351) > at > org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.java:259) > at > org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.java:250) > at > job.KafkaJoinHbase2Hbase.testJoinHbaseWithPrecision(KafkaJoinHbase2Hbase.java:109) > at job.KafkaJoinHbase2Hbase.main(KafkaJoinHbase2Hbase.java:24)Caused by: > java.lang.AssertionError: Type mismatch:rowtype of new > rel:RecordType(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" order_id, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" item, VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" currency, DECIMAL(10, 4) amount, TIMESTAMP(3) > order_time, TIME ATTRIBUTE(PROCTIME) NOT NULL proc_time, DECIMAL(20, 4) > amount_kg, TIME ATTRIBUTE(ROWTIME) ts, BIGINT currency_id, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency_name, DECIMAL(38, 4) > rate, TIMESTAMP(3) currency_time, VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" country, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" rowkey, > RecordType:peek_no_expand(INTEGER country_id, VARCHAR(2147483647) CHARACTER > SET "UTF-16LE" country_name, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > country_name_cn, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" region_name) f1, > RecordType:peek_no_expand(TIMESTAMP(3) record_timestamp3, TIMESTAMP(3) > record_timestamp9, TIME(0) time3, TIME(0) time9, DECIMAL(38, 18) gdp) f2) NOT > NULLrowtype of set:RecordType(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > order_id, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" item, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, DECIMAL(10, 4) amount, > TIMESTAMP(3) order_time, TIME ATTRIBUTE(PROCTIME) NOT NULL proc_time, > DECIMAL(20, 4) amount_kg, TIME ATTRIBUTE(ROWTIME) ts, BIGINT currency_id, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency_name, DECIMAL(38, 4) > rate, TIMESTAMP(3) currency_time, VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" country, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" rowkey, > RecordType:peek_no_expand(INTEGER country_id, VARCHAR(2147483647) CHARACTER > SET "UTF-16LE" country_name, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > country_name_cn, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, > VARCHAR(2147483647) CHARACTER SET "UTF-16LE" region_name) f1, > RecordType:peek_no_expand(TIMESTAMP(3) record_timestamp3, TIMESTAMP(9) > record_timestamp9, TIME(0) time3, TIME(0) time9, DECIMAL(10, 4) gdp) f2) NOT > NULL at org.apache.calcite.util.Litmus$1.fail(Litmus.java:31) at > org.apache.calcite.plan.RelOptUtil.equal(RelOptUtil.java:2026) at > org.apache.calcite.plan.volcano.RelSubset.add(RelSubset.java:284) at > org.apache.calcite.plan.volcano.RelSet.add(RelSet.java:148) at > org.apache.calcite.plan.volcano.VolcanoPlanner.addRelToSet(VolcanoPlanner.java:1818) > at > org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1764) > at > org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:846) > at > org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:868) > at > org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:1939) > at > org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:129) > at > org.apache.calcite.plan.RelOptRuleCall.transformTo(RelOptRuleCall.java:236) > at > org.apache.flink.table.planner.plan.rules.physical.common.BaseSnapshotOnTableScanRule.onMatch(CommonLookupJoinRule.scala:142) > at > org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:208) > at > org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:631) > at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:327) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:64) > ... 21 more > Process finished with exit code 1 > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)