You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/04/28 11:33:30 UTC
[flink-connector-hbase] branch main updated: [FLINK-30360][Connectors/HBase] When specified the partial columns, ignore all empty columns
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-hbase.git
The following commit(s) were added to refs/heads/main by this push:
new affc2db [FLINK-30360][Connectors/HBase] When specified the partial columns, ignore all empty columns
affc2db is described below
commit affc2dbbdb0ef1ef8a050a29944fb62292699dba
Author: Jichao Wang <wj...@foxmail.com>
AuthorDate: Fri Dec 16 20:51:38 2022 +0800
[FLINK-30360][Connectors/HBase] When specified the partial columns, ignore all empty columns
---
.../connector/hbase1/HBaseConnectorITCase.java | 55 ++++++++++++++++++++++
.../flink/connector/hbase1/util/HBaseTestBase.java | 8 ++++
.../connector/hbase2/HBaseConnectorITCase.java | 55 ++++++++++++++++++++++
.../flink/connector/hbase2/util/HBaseTestBase.java | 8 ++++
.../flink/connector/hbase/util/HBaseSerde.java | 3 ++
5 files changed, 129 insertions(+)
diff --git a/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java b/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
index 1ad2dd2..38046bd 100644
--- a/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
+++ b/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/HBaseConnectorITCase.java
@@ -280,6 +280,61 @@ public class HBaseConnectorITCase extends HBaseTestBase {
TestBaseUtils.compareResultAsText(results, expected);
}
+ @Test
+ public void testPartialColumnTableSink() throws Exception {
+ StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(execEnv, streamSettings);
+
+ // register HBase table testTable1 which contains test data
+ String table1DDL = createHBaseTableDDL(TEST_TABLE_1, false);
+ tEnv.executeSql(table1DDL);
+
+ String table2DDL = createHBaseTableDDL(TEST_TABLE_4, false);
+ tEnv.executeSql(table2DDL);
+
+ String query =
+ "INSERT INTO "
+ + TEST_TABLE_4
+ + " (rowkey, family3) "
+ + " SELECT"
+ + " rowkey,"
+ + " family3"
+ + " FROM "
+ + TEST_TABLE_1;
+
+ tEnv.executeSql(query).await();
+
+ // start a batch scan job to verify contents in HBase table
+ TableEnvironment batchEnv = TableEnvironment.create(batchSettings);
+ batchEnv.executeSql(table2DDL);
+
+ Table table =
+ batchEnv.sqlQuery(
+ "SELECT "
+ + " h.rowkey, "
+ + " h.family1.col1, "
+ + " h.family2.col1, "
+ + " h.family2.col2, "
+ + " h.family3.col1, "
+ + " h.family3.col2, "
+ + " h.family3.col3 "
+ + "FROM "
+ + TEST_TABLE_4
+ + " AS h");
+ List<Row> results = CollectionUtil.iteratorToList(table.execute().collect());
+ String expected =
+ "+I[1, null, null, null, 1.01, false, Welt-1]\n"
+ + "+I[2, null, null, null, 2.02, true, Welt-2]\n"
+ + "+I[3, null, null, null, 3.03, false, Welt-3]\n"
+ + "+I[4, null, null, null, 4.04, true, Welt-4]\n"
+ + "+I[5, null, null, null, 5.05, false, Welt-5]\n"
+ + "+I[6, null, null, null, 6.06, true, Welt-6]\n"
+ + "+I[7, null, null, null, 7.07, false, Welt-7]\n"
+ + "+I[8, null, null, null, 8.08, true, Welt-8]\n";
+
+ TestBaseUtils.compareResultAsText(results, expected);
+ }
+
@Test
public void testTableSourceSinkWithDDL() throws Exception {
StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java b/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
index cc2de79..7a589d4 100644
--- a/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
+++ b/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
@@ -43,6 +43,7 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
protected static final String TEST_TABLE_1 = "testTable1";
protected static final String TEST_TABLE_2 = "testTable2";
protected static final String TEST_TABLE_3 = "testTable3";
+ protected static final String TEST_TABLE_4 = "testTable4";
protected static final String ROW_KEY = "rowkey";
@@ -92,6 +93,7 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
createHBaseTable1();
createHBaseTable2();
createHBaseTable3();
+ createHBaseTable4();
}
private static void createHBaseTable1() throws IOException {
@@ -232,6 +234,12 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
createTable(tableName, families, SPLIT_KEYS);
}
+ private static void createHBaseTable4() {
+ // create a table
+ TableName tableName = TableName.valueOf(TEST_TABLE_4);
+ createTable(tableName, FAMILIES, SPLIT_KEYS);
+ }
+
private static Put putRow(
int rowKey,
int f1c1,
diff --git a/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java b/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java
index e75d0d0..572b876 100644
--- a/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java
+++ b/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/HBaseConnectorITCase.java
@@ -311,6 +311,61 @@ public class HBaseConnectorITCase extends HBaseTestBase {
TestBaseUtils.compareResultAsText(results, String.join("", expected));
}
+ @Test
+ public void testPartialColumnTableSink() throws Exception {
+ StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(execEnv, streamSettings);
+
+ // register HBase table testTable1 which contains test data
+ String table1DDL = createHBaseTableDDL(TEST_TABLE_1, false);
+ tEnv.executeSql(table1DDL);
+
+ String table2DDL = createHBaseTableDDL(TEST_TABLE_4, false);
+ tEnv.executeSql(table2DDL);
+
+ String query =
+ "INSERT INTO "
+ + TEST_TABLE_4
+ + " (rowkey, family3) "
+ + " SELECT"
+ + " rowkey,"
+ + " family3"
+ + " FROM "
+ + TEST_TABLE_1;
+
+ tEnv.executeSql(query).await();
+
+ // start a batch scan job to verify contents in HBase table
+ TableEnvironment batchEnv = TableEnvironment.create(batchSettings);
+ batchEnv.executeSql(table2DDL);
+
+ Table table =
+ batchEnv.sqlQuery(
+ "SELECT "
+ + " h.rowkey, "
+ + " h.family1.col1, "
+ + " h.family2.col1, "
+ + " h.family2.col2, "
+ + " h.family3.col1, "
+ + " h.family3.col2, "
+ + " h.family3.col3 "
+ + "FROM "
+ + TEST_TABLE_4
+ + " AS h");
+ List<Row> results = CollectionUtil.iteratorToList(table.execute().collect());
+ String expected =
+ "+I[1, null, null, null, 1.01, false, Welt-1]\n"
+ + "+I[2, null, null, null, 2.02, true, Welt-2]\n"
+ + "+I[3, null, null, null, 3.03, false, Welt-3]\n"
+ + "+I[4, null, null, null, 4.04, true, Welt-4]\n"
+ + "+I[5, null, null, null, 5.05, false, Welt-5]\n"
+ + "+I[6, null, null, null, 6.06, true, Welt-6]\n"
+ + "+I[7, null, null, null, 7.07, false, Welt-7]\n"
+ + "+I[8, null, null, null, 8.08, true, Welt-8]\n";
+
+ TestBaseUtils.compareResultAsText(results, expected);
+ }
+
@Test
public void testTableSourceSinkWithDDL() throws Exception {
StreamExecutionEnvironment execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java b/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
index 1e639ba..331aabf 100644
--- a/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
+++ b/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
@@ -43,6 +43,7 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
protected static final String TEST_TABLE_1 = "testTable1";
protected static final String TEST_TABLE_2 = "testTable2";
protected static final String TEST_TABLE_3 = "testTable3";
+ protected static final String TEST_TABLE_4 = "testTable4";
protected static final String ROW_KEY = "rowkey";
@@ -92,6 +93,7 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
createHBaseTable1();
createHBaseTable2();
createHBaseTable3();
+ createHBaseTable4();
}
private static void createHBaseTable1() throws IOException {
@@ -232,6 +234,12 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
createTable(tableName, families, SPLIT_KEYS);
}
+ private static void createHBaseTable4() {
+ // create a table
+ TableName tableName = TableName.valueOf(TEST_TABLE_4);
+ createTable(tableName, FAMILIES, SPLIT_KEYS);
+ }
+
private static Put putRow(
int rowKey,
int f1c1,
diff --git a/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseSerde.java b/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseSerde.java
index d21cc4a..a8863ff 100644
--- a/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseSerde.java
+++ b/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseSerde.java
@@ -140,6 +140,9 @@ public class HBaseSerde {
// get family key
byte[] familyKey = families[f];
RowData familyRow = row.getRow(i, qualifiers[f].length);
+ if (familyRow == null) {
+ continue;
+ }
for (int q = 0; q < this.qualifiers[f].length; q++) {
// get quantifier key
byte[] qualifier = qualifiers[f][q];