wuchong commented on a change in pull request #12594:
URL: https://github.com/apache/flink/pull/12594#discussion_r440262864
##########
File path:
flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowDataLookupFunction.java
##########
@@ -79,7 +77,7 @@ public HBaseRowDataLookupFunction(
*/
public void eval(Object rowKey) throws IOException {
// fetch result
- Result result = table.get(readHelper.createGet(rowKey));
+ Result result = table.get(serde.createGet(rowKey));
Review comment:
The returned Get may be null.
##########
File path:
flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/util/HBaseSerde.java
##########
@@ -190,6 +191,30 @@ public Scan createScan() {
return scan;
}
+ /**
+ * Returns an instance of Get that retrieves the matches records from
the HBase table.
+ *
+ * @return The appropriate instance of Get for this use case.
+ */
+ public Get createGet(Object rowKey) {
+ checkArgument(keyEncoder != null, "row key is not set.");
+ GenericRowData rowData = new GenericRowData(1);
Review comment:
We can reuse the `GenericRowData`.
##########
File path:
flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java
##########
@@ -455,99 +392,82 @@ public void testTableSink() throws Exception {
@Test
public void testTableSourceSinkWithDDL() throws Exception {
+ // only test TIMESTAMP/DATE/TIME/DECIMAL for new
connector(using blink-planner), because new connector encodes
+ // DATE/TIME to int, the old one encodes to long, and DECIMAL
with precision works well in new connector.
+ final boolean testTimeAndDecimalTypes =
BLINK_PLANNER.equals(planner) && !isLegacyConnector;
Review comment:
This test code is really hard to maintain and read. I would suggest only
test for blink planner and new connector for simplification.
##########
File path:
flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java
##########
@@ -377,57 +344,28 @@ public void testTableSink() throws Exception {
TableSink tableSink = TableFactoryService
.find(HBaseTableFactory.class,
descriptorProperties.asMap())
.createTableSink(descriptorProperties.asMap());
- ((TableEnvironmentInternal)
tEnv).registerTableSinkInternal("hbase", tableSink);
+ ((TableEnvironmentInternal)
tEnv).registerTableSinkInternal(TEST_TABLE_2, tableSink);
Review comment:
Can we also migrate this to use DDL?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
