This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 422ea607 [FLINK-29278] BINARY type is not supported in table store
422ea607 is described below
commit 422ea6072b251b04041b6ca8a738316a30069aad
Author: Nicholas Jiang <[email protected]>
AuthorDate: Wed Sep 21 19:21:03 2022 +0800
[FLINK-29278] BINARY type is not supported in table store
This closes #292
---
.../store/format/FileStatsExtractorTestBase.java | 8 ++
.../store/table/AppendOnlyFileStoreTableTest.java | 51 ++++++----
.../ChangelogValueCountFileStoreTableTest.java | 56 ++++++----
.../table/ChangelogWithKeyFileStoreTableTest.java | 113 ++++++++++++---------
.../table/store/table/FileStoreTableTestBase.java | 77 +++++++++-----
.../table/store/table/WritePreemptMemoryTest.java | 3 +-
.../table/store/format/orc/OrcFileFormat.java | 45 +++++++-
.../format/orc/OrcFileStatsExtractorTest.java | 4 +
8 files changed, 235 insertions(+), 122 deletions(-)
diff --git
a/flink-table-store-common/src/test/java/org/apache/flink/table/store/format/FileStatsExtractorTestBase.java
b/flink-table-store-common/src/test/java/org/apache/flink/table/store/format/FileStatsExtractorTestBase.java
index 90a53f7f..18d5c89a 100644
---
a/flink-table-store-common/src/test/java/org/apache/flink/table/store/format/FileStatsExtractorTestBase.java
+++
b/flink-table-store-common/src/test/java/org/apache/flink/table/store/format/FileStatsExtractorTestBase.java
@@ -30,11 +30,13 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.CharType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.junit.jupiter.api.Test;
@@ -128,6 +130,12 @@ public abstract class FileStatsExtractorTestBase {
randomString(random.nextInt(varCharType.getLength()) +
1));
case BOOLEAN:
return random.nextBoolean();
+ case BINARY:
+ BinaryType binaryType = (BinaryType) type;
+ return randomString(binaryType.getLength()).getBytes();
+ case VARBINARY:
+ VarBinaryType varBinaryType = (VarBinaryType) type;
+ return randomString(varBinaryType.getLength()).getBytes();
case TINYINT:
return (byte) random.nextInt(10);
case SMALLINT:
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
index 620252bb..072eec88 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.table;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
@@ -63,9 +62,19 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
.hasSameElementsAs(
- Arrays.asList("1|10|100", "1|11|101", "1|12|102",
"1|11|101", "1|12|102"));
+ Arrays.asList(
+ "1|10|100|binary|varbinary",
+ "1|11|101|binary|varbinary",
+ "1|12|102|binary|varbinary",
+ "1|11|101|binary|varbinary",
+ "1|12|102|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Arrays.asList("2|20|200", "2|21|201",
"2|22|202", "2|21|201"));
+ .hasSameElementsAs(
+ Arrays.asList(
+ "2|20|200|binary|varbinary",
+ "2|21|201|binary|varbinary",
+ "2|22|202|binary|varbinary",
+ "2|21|201|binary|varbinary"));
}
@Test
@@ -94,10 +103,10 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
.hasSameElementsAs(
Arrays.asList(
- "2|21|201",
+ "2|21|201|binary|varbinary",
// this record is in the same file with the
first "2|21|201"
- "2|22|202",
- "2|21|201"));
+ "2|22|202|binary|varbinary",
+ "2|21|201|binary|varbinary"));
}
@Test
@@ -108,9 +117,10 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
List<Split> splits =
table.newScan().withIncremental(true).plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("+1|11|101", "+1|12|102"));
+ .isEqualTo(
+ Arrays.asList("+1|11|101|binary|varbinary",
"+1|12|102|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Collections.singletonList("+2|21|201"));
+
.isEqualTo(Collections.singletonList("+2|21|201|binary|varbinary"));
}
@Test
@@ -139,9 +149,9 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
assertThat(getResult(read, splits, binaryRow(1), 0,
STREAMING_ROW_TO_STRING))
.isEqualTo(
Arrays.asList(
- "+1|11|101",
+ "+1|11|101|binary|varbinary",
// this record is in the same file with
"+1|11|101"
- "+1|12|102"));
+ "+1|12|102|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
STREAMING_ROW_TO_STRING)).isEmpty();
}
@@ -161,8 +171,7 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
for (int j = 0; j < Math.max(random.nextInt(200), 1); j++) {
BinaryRowData data =
serializer
- .toBinaryRow(
- GenericRowData.of(i, random.nextInt(),
random.nextLong()))
+ .toBinaryRow(rowData(i, random.nextInt(),
random.nextLong()))
.copy();
int bucket = bucket(hashcode(data), numOfBucket);
dataPerBucket.compute(
@@ -203,19 +212,19 @@ public class AppendOnlyFileStoreTableTest extends
FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(2, 20, 200L));
- write.write(GenericRowData.of(1, 11, 101L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(2, 20, 200L));
+ write.write(rowData(1, 11, 101L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 12, 102L));
- write.write(GenericRowData.of(2, 21, 201L));
- write.write(GenericRowData.of(2, 22, 202L));
+ write.write(rowData(1, 12, 102L));
+ write.write(rowData(2, 21, 201L));
+ write.write(rowData(2, 22, 202L));
commit.commit("1", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 11, 101L));
- write.write(GenericRowData.of(2, 21, 201L));
- write.write(GenericRowData.of(1, 12, 102L));
+ write.write(rowData(1, 11, 101L));
+ write.write(rowData(2, 21, 201L));
+ write.write(rowData(1, 12, 102L));
commit.commit("2", write.prepareCommit(true));
write.close();
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
index 894f5d85..8ce299dd 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.table;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.file.WriteMode;
import org.apache.flink.table.store.file.predicate.Predicate;
@@ -53,9 +52,17 @@ public class ChangelogValueCountFileStoreTableTest extends
FileStoreTableTestBas
List<Split> splits = table.newScan().plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("1|11|101", "1|11|101", "1|12|102"));
+ .isEqualTo(
+ Arrays.asList(
+ "1|11|101|binary|varbinary",
+ "1|11|101|binary|varbinary",
+ "1|12|102|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("2|20|200", "2|21|201", "2|22|202"));
+ .isEqualTo(
+ Arrays.asList(
+ "2|20|200|binary|varbinary",
+ "2|21|201|binary|varbinary",
+ "2|22|202|binary|varbinary"));
}
@Test
@@ -84,9 +91,9 @@ public class ChangelogValueCountFileStoreTableTest extends
FileStoreTableTestBas
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
.isEqualTo(
Arrays.asList(
- "2|21|201",
+ "2|21|201|binary|varbinary",
// this record is in the same file with
"delete 2|21|201"
- "2|22|202"));
+ "2|22|202|binary|varbinary"));
}
@Test
@@ -97,9 +104,14 @@ public class ChangelogValueCountFileStoreTableTest extends
FileStoreTableTestBas
List<Split> splits =
table.newScan().withIncremental(true).plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("-1|10|100", "+1|11|101"));
+ .isEqualTo(
+ Arrays.asList("-1|10|100|binary|varbinary",
"+1|11|101|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("-2|21|201", "-2|21|201",
"+2|22|202"));
+ .isEqualTo(
+ Arrays.asList(
+ "-2|21|201|binary|varbinary",
+ "-2|21|201|binary|varbinary",
+ "+2|22|202|binary|varbinary"));
}
@Test
@@ -129,10 +141,10 @@ public class ChangelogValueCountFileStoreTableTest
extends FileStoreTableTestBas
assertThat(getResult(read, splits, binaryRow(2), 0,
STREAMING_ROW_TO_STRING))
.isEqualTo(
Arrays.asList(
- "-2|21|201",
- "-2|21|201",
+ "-2|21|201|binary|varbinary",
+ "-2|21|201|binary|varbinary",
// this record is in the same file with
"delete 2|21|201"
- "+2|22|202"));
+ "+2|22|202|binary|varbinary"));
}
private void writeData() throws Exception {
@@ -140,22 +152,22 @@ public class ChangelogValueCountFileStoreTableTest
extends FileStoreTableTestBas
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(2, 20, 200L));
- write.write(GenericRowData.of(1, 11, 101L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(2, 20, 200L));
+ write.write(rowData(1, 11, 101L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(2, 21, 201L));
- write.write(GenericRowData.of(1, 12, 102L));
- write.write(GenericRowData.of(2, 21, 201L));
- write.write(GenericRowData.of(2, 21, 201L));
+ write.write(rowData(2, 21, 201L));
+ write.write(rowData(1, 12, 102L));
+ write.write(rowData(2, 21, 201L));
+ write.write(rowData(2, 21, 201L));
commit.commit("1", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 11, 101L));
- write.write(GenericRowData.of(2, 22, 202L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 2, 21, 201L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 1, 10, 100L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 2, 21, 201L));
+ write.write(rowData(1, 11, 101L));
+ write.write(rowData(2, 22, 202L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 21, 201L));
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 21, 201L));
commit.commit("2", write.prepareCommit(true));
write.close();
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index 8ff0b786..3f4e908d 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.table;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.CoreOptions.ChangelogProducer;
import org.apache.flink.table.store.file.WriteMode;
@@ -52,18 +51,18 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
createFileStoreTable(conf ->
conf.set(CoreOptions.SEQUENCE_FIELD, "b"));
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 200L));
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(1, 11, 101L));
+ write.write(rowData(1, 10, 200L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(1, 11, 101L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 11, 55L));
+ write.write(rowData(1, 11, 55L));
commit.commit("1", write.prepareCommit(true));
write.close();
List<Split> splits = table.newScan().plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("1|10|200", "1|11|101"));
+ .isEqualTo(Arrays.asList("1|10|200|binary|varbinary",
"1|11|101|binary|varbinary"));
}
@Test
@@ -74,9 +73,10 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
List<Split> splits = table.newScan().plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .isEqualTo(Collections.singletonList("1|10|1000"));
+
.isEqualTo(Collections.singletonList("1|10|1000|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("2|21|20001", "2|22|202"));
+ .isEqualTo(
+ Arrays.asList("2|21|20001|binary|varbinary",
"2|22|202|binary|varbinary"));
}
@Test
@@ -107,7 +107,7 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
Arrays.asList(
// only filter on key should be performed,
// and records from the same file should also
be selected
- "2|21|20001", "2|22|202"));
+ "2|21|20001|binary|varbinary",
"2|22|202|binary|varbinary"));
}
@Test
@@ -118,9 +118,13 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
List<Split> splits =
table.newScan().withIncremental(true).plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Collections.singletonList("-1|11|1001"));
+
.isEqualTo(Collections.singletonList("-1|11|1001|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
STREAMING_ROW_TO_STRING))
- .isEqualTo(Arrays.asList("-2|20|200", "+2|21|20001",
"+2|22|202"));
+ .isEqualTo(
+ Arrays.asList(
+ "-2|20|200|binary|varbinary",
+ "+2|21|20001|binary|varbinary",
+ "+2|22|202|binary|varbinary"));
}
@Test
@@ -153,7 +157,9 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
Arrays.asList(
// only filter on key should be performed,
// and records from the same file should also
be selected
- "-2|20|200", "+2|21|20001", "+2|22|202"));
+ "-2|20|200|binary|varbinary",
+ "+2|21|20001|binary|varbinary",
+ "+2|22|202|binary|varbinary"));
}
@Test
@@ -163,11 +169,11 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
conf -> conf.set(CoreOptions.CHANGELOG_PRODUCER,
ChangelogProducer.INPUT));
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 1, 10, 100L));
- write.write(GenericRowData.of(1, 10, 101L));
- write.write(GenericRowData.ofKind(RowKind.UPDATE_BEFORE, 1, 10, 101L));
- write.write(GenericRowData.ofKind(RowKind.UPDATE_AFTER, 1, 10, 102L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
+ write.write(rowData(1, 10, 101L));
+ write.write(rowDataWithKind(RowKind.UPDATE_BEFORE, 1, 10, 101L));
+ write.write(rowDataWithKind(RowKind.UPDATE_AFTER, 1, 10, 102L));
commit.commit("0", write.prepareCommit(true));
write.close();
@@ -176,11 +182,11 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
assertThat(getResult(read, splits, binaryRow(1), 0,
CHANGELOG_ROW_TO_STRING))
.isEqualTo(
Arrays.asList(
- "+I 1|10|100",
- "-D 1|10|100",
- "+I 1|10|101",
- "-U 1|10|101",
- "+U 1|10|102"));
+ "+I 1|10|100|binary|varbinary",
+ "-D 1|10|100|binary|varbinary",
+ "+I 1|10|101|binary|varbinary",
+ "-U 1|10|101|binary|varbinary",
+ "+U 1|10|102|binary|varbinary"));
}
private void writeData() throws Exception {
@@ -188,21 +194,21 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(2, 20, 200L));
- write.write(GenericRowData.of(1, 11, 101L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(2, 20, 200L));
+ write.write(rowData(1, 11, 101L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 10, 1000L));
- write.write(GenericRowData.of(2, 21, 201L));
- write.write(GenericRowData.of(2, 21, 2001L));
+ write.write(rowData(1, 10, 1000L));
+ write.write(rowData(2, 21, 201L));
+ write.write(rowData(2, 21, 2001L));
commit.commit("1", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 11, 1001L));
- write.write(GenericRowData.of(2, 21, 20001L));
- write.write(GenericRowData.of(2, 22, 202L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 1, 11, 1001L));
- write.write(GenericRowData.ofKind(RowKind.DELETE, 2, 20, 200L));
+ write.write(rowData(1, 11, 1001L));
+ write.write(rowData(2, 21, 20001L));
+ write.write(rowData(2, 22, 202L));
+ write.write(rowDataWithKind(RowKind.DELETE, 1, 11, 1001L));
+ write.write(rowDataWithKind(RowKind.DELETE, 2, 20, 200L));
commit.commit("2", write.prepareCommit(true));
write.close();
@@ -216,16 +222,16 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(1, 20, 200L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(1, 20, 200L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 30, 300L));
- write.write(GenericRowData.of(1, 40, 400L));
+ write.write(rowData(1, 30, 300L));
+ write.write(rowData(1, 40, 400L));
commit.commit("1", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 50, 500L));
- write.write(GenericRowData.of(1, 60, 600L));
+ write.write(rowData(1, 50, 500L));
+ write.write(rowData(1, 60, 600L));
commit.commit("2", write.prepareCommit(true));
PredicateBuilder builder = new PredicateBuilder(ROW_TYPE);
@@ -234,12 +240,14 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
// push down key filter a = 30
TableRead read = table.newRead().withFilter(builder.equal(1, 30));
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Arrays.asList("1|30|300", "1|40|400"));
+ .hasSameElementsAs(
+ Arrays.asList("1|30|300|binary|varbinary",
"1|40|400|binary|varbinary"));
// push down value filter b = 300L
read = table.newRead().withFilter(builder.equal(2, 300L));
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Arrays.asList("1|30|300", "1|40|400"));
+ .hasSameElementsAs(
+ Arrays.asList("1|30|300|binary|varbinary",
"1|40|400|binary|varbinary"));
// push down both key filter and value filter
read =
@@ -247,11 +255,16 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
.withFilter(
PredicateBuilder.or(builder.equal(1, 10),
builder.equal(2, 300L)));
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Arrays.asList("1|10|100", "1|20|200",
"1|30|300", "1|40|400"));
+ .hasSameElementsAs(
+ Arrays.asList(
+ "1|10|100|binary|varbinary",
+ "1|20|200|binary|varbinary",
+ "1|30|300|binary|varbinary",
+ "1|40|400|binary|varbinary"));
// update pk 60, 10
- write.write(GenericRowData.of(1, 60, 500L));
- write.write(GenericRowData.of(1, 10, 10L));
+ write.write(rowData(1, 60, 500L));
+ write.write(rowData(1, 10, 10L));
commit.commit("3", write.prepareCommit(true));
write.close();
@@ -262,12 +275,12 @@ public class ChangelogWithKeyFileStoreTableTest extends
FileStoreTableTestBase {
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
.hasSameElementsAs(
Arrays.asList(
- "1|10|10",
- "1|20|200",
- "1|30|300",
- "1|40|400",
- "1|50|500",
- "1|60|500"));
+ "1|10|10|binary|varbinary",
+ "1|20|200|binary|varbinary",
+ "1|30|300|binary|varbinary",
+ "1|40|400|binary|varbinary",
+ "1|50|500|binary|varbinary",
+ "1|60|500|binary|varbinary"));
}
@Override
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
index 3c80b563..9a9c2087 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
@@ -71,12 +71,23 @@ public abstract class FileStoreTableTestBase {
new LogicalType[] {
DataTypes.INT().getLogicalType(),
DataTypes.INT().getLogicalType(),
- DataTypes.BIGINT().getLogicalType()
+ DataTypes.BIGINT().getLogicalType(),
+ DataTypes.BINARY(1).getLogicalType(),
+ DataTypes.VARBINARY(1).getLogicalType()
},
- new String[] {"pt", "a", "b"});
+ new String[] {"pt", "a", "b", "c", "d"});
protected static final int[] PROJECTION = new int[] {2, 1};
protected static final Function<RowData, String> BATCH_ROW_TO_STRING =
- rowData -> rowData.getInt(0) + "|" + rowData.getInt(1) + "|" +
rowData.getLong(2);
+ rowData ->
+ rowData.getInt(0)
+ + "|"
+ + rowData.getInt(1)
+ + "|"
+ + rowData.getLong(2)
+ + "|"
+ + new String(rowData.getBinary(3))
+ + "|"
+ + new String(rowData.getBinary(4));
protected static final Function<RowData, String>
BATCH_PROJECTED_ROW_TO_STRING =
rowData -> rowData.getLong(0) + "|" + rowData.getInt(1);
protected static final Function<RowData, String> STREAMING_ROW_TO_STRING =
@@ -106,14 +117,14 @@ public abstract class FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(2, 20, 200L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(2, 20, 200L));
commit.commit("0", write.prepareCommit(true));
write.close();
write = table.newWrite().withOverwrite(true);
commit = table.newCommit("user");
- write.write(GenericRowData.of(2, 21, 201L));
+ write.write(rowData(2, 21, 201L));
Map<String, String> overwritePartition = new HashMap<>();
overwritePartition.put("pt", "2");
commit.withOverwritePartition(overwritePartition).commit("1",
write.prepareCommit(true));
@@ -122,9 +133,9 @@ public abstract class FileStoreTableTestBase {
List<Split> splits = table.newScan().plan().splits;
TableRead read = table.newRead();
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Collections.singletonList("1|10|100"));
+
.hasSameElementsAs(Collections.singletonList("1|10|100|binary|varbinary"));
assertThat(getResult(read, splits, binaryRow(2), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Collections.singletonList("2|21|201"));
+
.hasSameElementsAs(Collections.singletonList("2|21|201|binary|varbinary"));
}
@Test
@@ -137,11 +148,11 @@ public abstract class FileStoreTableTestBase {
});
TableWrite write = table.newWrite();
- write.write(GenericRowData.of(1, 1, 2L));
- write.write(GenericRowData.of(1, 3, 4L));
- write.write(GenericRowData.of(1, 5, 6L));
- write.write(GenericRowData.of(1, 7, 8L));
- write.write(GenericRowData.of(1, 9, 10L));
+ write.write(rowData(1, 1, 2L));
+ write.write(rowData(1, 3, 4L));
+ write.write(rowData(1, 5, 6L));
+ write.write(rowData(1, 7, 8L));
+ write.write(rowData(1, 9, 10L));
table.newCommit("user").commit("0", write.prepareCommit(true));
write.close();
@@ -161,16 +172,16 @@ public abstract class FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
- write.write(GenericRowData.of(1, 10, 100L));
- write.write(GenericRowData.of(1, 20, 200L));
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(1, 20, 200L));
commit.commit("0", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 30, 300L));
- write.write(GenericRowData.of(1, 40, 400L));
+ write.write(rowData(1, 30, 300L));
+ write.write(rowData(1, 40, 400L));
commit.commit("1", write.prepareCommit(true));
- write.write(GenericRowData.of(1, 50, 500L));
- write.write(GenericRowData.of(1, 60, 600L));
+ write.write(rowData(1, 50, 500L));
+ write.write(rowData(1, 60, 600L));
commit.commit("2", write.prepareCommit(true));
write.close();
@@ -179,7 +190,8 @@ public abstract class FileStoreTableTestBase {
List<Split> splits = table.newScan().plan().splits;
TableRead read = table.newRead().withFilter(builder.equal(2, 300L));
assertThat(getResult(read, splits, binaryRow(1), 0,
BATCH_ROW_TO_STRING))
- .hasSameElementsAs(Arrays.asList("1|30|300", "1|40|400"));
+ .hasSameElementsAs(
+ Arrays.asList("1|30|300|binary|varbinary",
"1|40|400|binary|varbinary"));
}
@Test
@@ -191,21 +203,21 @@ public abstract class FileStoreTableTestBase {
for (int i = 0; i < 4; i++) {
// write lots of records, let compaction be slower
for (int j = 0; j < 1000; j++) {
- write.write(GenericRowData.of(1, 10 * i * j, 100L * i * j));
+ write.write(rowData(1, 10 * i * j, 100L * i * j));
}
commit.commit(String.valueOf(i), write.prepareCommit(false));
}
- write.write(GenericRowData.of(1, 40, 400L));
+ write.write(rowData(1, 40, 400L));
List<FileCommittable> commit4 = write.prepareCommit(false);
// trigger compaction, but not wait it.
- write.write(GenericRowData.of(2, 20, 200L));
+ write.write(rowData(2, 20, 200L));
List<FileCommittable> commit5 = write.prepareCommit(true);
// wait compaction finish
// commit5 should be a compaction commit
- write.write(GenericRowData.of(1, 60, 600L));
+ write.write(rowData(1, 60, 600L));
List<FileCommittable> commit6 = write.prepareCommit(true);
// if remove writer too fast, will see old files, do another compaction
// then will be conflicts
@@ -229,7 +241,7 @@ public abstract class FileStoreTableTestBase {
TableWrite write = table.newWrite();
TableCommit commit = table.newCommit("user");
for (int i = 0; i < 10; i++) {
- write.write(GenericRowData.of(1, 1, 100L));
+ write.write(rowData(1, 1, 100L));
commit.commit(String.valueOf(i), write.prepareCommit(true));
}
write.close();
@@ -291,6 +303,21 @@ public abstract class FileStoreTableTestBase {
return b;
}
+ protected GenericRowData rowData(Object... values) {
+ return GenericRowData.of(
+ values[0], values[1], values[2], "binary".getBytes(),
"varbinary".getBytes());
+ }
+
+ protected GenericRowData rowDataWithKind(RowKind rowKind, Object...
values) {
+ return GenericRowData.ofKind(
+ rowKind,
+ values[0],
+ values[1],
+ values[2],
+ "binary".getBytes(),
+ "varbinary".getBytes());
+ }
+
protected FileStoreTable createFileStoreTable(int numOfBucket) throws
Exception {
return createFileStoreTable(conf -> conf.set(BUCKET, numOfBucket));
}
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
index 7ddb76dd..353ac24e 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
@@ -68,8 +68,7 @@ public class WritePreemptMemoryTest extends
FileStoreTableTestBase {
Random random = new Random();
List<String> expected = new ArrayList<>();
for (int i = 0; i < 10_000; i++) {
- GenericRowData row =
- GenericRowData.of(singlePartition ? 0 : random.nextInt(5),
i, i * 10L);
+ GenericRowData row = rowData(singlePartition ? 0 :
random.nextInt(5), i, i * 10L);
write.write(row);
expected.add(BATCH_ROW_TO_STRING.apply(row));
}
diff --git
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
index 9910409e..20003de0 100644
---
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
+++
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
@@ -29,12 +29,15 @@ import org.apache.flink.orc.OrcSplitReaderUtil;
import org.apache.flink.orc.vector.RowDataVectorizer;
import org.apache.flink.orc.vector.Vectorizer;
import org.apache.flink.orc.writer.ThreadLocalClassLoaderConfiguration;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.format.FileStatsExtractor;
import org.apache.flink.table.store.utils.Projection;
+import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.orc.OrcFile;
@@ -46,6 +49,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
+import java.util.stream.Collectors;
import static
org.apache.flink.table.store.format.orc.OrcFileFormatFactory.IDENTIFIER;
@@ -92,14 +96,18 @@ public class OrcFileFormat extends FileFormat {
}
return OrcInputFormatFactory.create(
- readerConf, type,
Projection.of(projection).toTopLevelIndexes(), orcPredicates);
+ readerConf,
+ (RowType) refineLogicalType(type),
+ Projection.of(projection).toTopLevelIndexes(),
+ orcPredicates);
}
@Override
public BulkWriter.Factory<RowData> createWriterFactory(RowType type) {
LogicalType[] orcTypes = type.getChildren().toArray(new
LogicalType[0]);
- TypeDescription typeDescription =
OrcSplitReaderUtil.logicalTypeToOrcType(type);
+ TypeDescription typeDescription =
+
OrcSplitReaderUtil.logicalTypeToOrcType(refineLogicalType(type));
Vectorizer<RowData> vectorizer =
new RowDataVectorizer(typeDescription.toString(), orcTypes);
OrcFile.WriterOptions writerOptions =
OrcFile.writerOptions(orcProperties, writerConf);
@@ -115,4 +123,37 @@ public class OrcFileFormat extends FileFormat {
properties.forEach((k, v) -> orcProperties.put(IDENTIFIER + "." + k,
v));
return orcProperties;
}
+
+ private static LogicalType refineLogicalType(LogicalType type) {
+ switch (type.getTypeRoot()) {
+ case BINARY:
+ case VARBINARY:
+ // OrcSplitReaderUtil#logicalTypeToOrcType() only supports the
DataTypes.BYTES()
+ // logical type for BINARY and VARBINARY.
+ return DataTypes.BYTES().getLogicalType();
+ case ARRAY:
+ ArrayType arrayType = (ArrayType) type;
+ return new ArrayType(
+ arrayType.isNullable(),
refineLogicalType(arrayType.getElementType()));
+ case MAP:
+ MapType mapType = (MapType) type;
+ return new MapType(
+ refineLogicalType(mapType.getKeyType()),
+ refineLogicalType(mapType.getValueType()));
+ case ROW:
+ RowType rowType = (RowType) type;
+ return new RowType(
+ rowType.isNullable(),
+ rowType.getFields().stream()
+ .map(
+ f ->
+ new RowType.RowField(
+ f.getName(),
+
refineLogicalType(f.getType()),
+
f.getDescription().orElse(null)))
+ .collect(Collectors.toList()));
+ default:
+ return type;
+ }
+ }
}
diff --git
a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileStatsExtractorTest.java
b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileStatsExtractorTest.java
index eaac4b29..86aba5c4 100644
---
a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileStatsExtractorTest.java
+++
b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileStatsExtractorTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.store.format.FileFormat;
import org.apache.flink.table.store.format.FileStatsExtractorTestBase;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.BooleanType;
import org.apache.flink.table.types.logical.CharType;
import org.apache.flink.table.types.logical.DateType;
@@ -34,6 +35,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.SmallIntType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
/** Tests for {@link OrcFileStatsExtractor}. */
@@ -50,6 +52,8 @@ public class OrcFileStatsExtractorTest extends
FileStatsExtractorTestBase {
new CharType(8),
new VarCharType(8),
new BooleanType(),
+ new BinaryType(8),
+ new VarBinaryType(8),
new TinyIntType(),
new SmallIntType(),
new IntType(),
