You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lsyldliu (via GitHub)" <gi...@apache.org> on 2023/06/29 16:33:29 UTC
[GitHub] [flink] lsyldliu opened a new pull request, #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
lsyldliu opened a new pull request, #22908:
URL: https://github.com/apache/flink/pull/22908
## What is the purpose of the change
*Runtime filter needs to use BloomFIlter, it would be transferred from runtime filter builder operator to runtime filter operator via network, so serialization and merge need.*
## Brief change log
- *Support serialize and merge BloomFilter for runtime filter*
## Verifying this change
This change added tests and can be verified as follows:
- *Added unit test in BloomFIlterTest*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): ( no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: ( no)
- The S3 file system connector: ( no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (JavaDocs)
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22908:
URL: https://github.com/apache/flink/pull/22908#issuecomment-1613511155
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5723ee7f7c47e74253617031e13e44ba0f52a08f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5723ee7f7c47e74253617031e13e44ba0f52a08f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5723ee7f7c47e74253617031e13e44ba0f52a08f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on PR #22908:
URL: https://github.com/apache/flink/pull/22908#issuecomment-1637020143
@wanglijie95 Thanks for your review, I've updated it.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #22908:
URL: https://github.com/apache/flink/pull/22908#discussion_r1264637661
##########
flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/BloomFilterTest.java:
##########
@@ -168,4 +173,161 @@ public void testHashcodeInput() {
assertTrue(bloomFilter.testHash(val4));
assertTrue(bloomFilter.testHash(val5));
}
+
+ @Test
+ public void testBloomFilterSerDe() {
Review Comment:
Yes, I have migrated all tests to junit5
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wanglijie95 commented on a diff in pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "wanglijie95 (via GitHub)" <gi...@apache.org>.
wanglijie95 commented on code in PR #22908:
URL: https://github.com/apache/flink/pull/22908#discussion_r1262051770
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BitSet.java:
##########
@@ -83,6 +83,13 @@ public int bitSize() {
return bitLength;
}
+ public byte[] getData() {
Review Comment:
Maybe `getData` -> `toBytes` or `toByteArray`, and add doc to indicate that currently only supports heap.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BitSet.java:
##########
@@ -83,6 +83,13 @@ public int bitSize() {
return bitLength;
}
+ public byte[] getData() {
+ if (memorySegment.isOffHeap()) {
+ throw new IllegalStateException("Only support use heap memory for serialization");
+ }
Review Comment:
This `if` block can repalced by the `checkState`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -135,6 +144,30 @@ public void reset() {
this.bitSet.clear();
}
+ /**
+ * Merge the specified bloom filter with current bloom filter.
+ *
+ * @param that - bloom filter to merge
+ */
+ public void merge(BloomFilter that) {
Review Comment:
Is this method needed? I think we only need the `mergeBloomFilterBytes`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -143,4 +176,63 @@ public String toString() {
output.append(bitSet);
return output.toString();
}
+
+ public static byte[] toBytes(BloomFilter filter) {
+ byte[] data = filter.bitSet.getData();
+ int byteSize = data.length;
+ byte[] bytes = new byte[8 + byteSize];
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET, filter.numHashFunctions);
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET + 4, byteSize);
+ UNSAFE.copyMemory(
+ data, BYTE_ARRAY_BASE_OFFSET, bytes, BYTE_ARRAY_BASE_OFFSET + 8, byteSize);
+ return bytes;
+ }
+
+ public static BloomFilter fromBytes(byte[] bytes) {
Review Comment:
Add doc, and indicate that currently only supports heap.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -43,17 +45,24 @@
*/
public class BloomFilter {
+ @SuppressWarnings("restriction")
+ private static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+
protected BitSet bitSet;
- protected int expectedEntries;
protected int numHashFunctions;
public BloomFilter(int expectedEntries, int byteSize) {
checkArgument(expectedEntries > 0, "expectedEntries should be > 0");
- this.expectedEntries = expectedEntries;
this.numHashFunctions = optimalNumOfHashFunctions(expectedEntries, byteSize << 3);
this.bitSet = new BitSet(byteSize);
}
+ /** A constructor to support rebuilding the BloomFilter from a serialized representation. */
+ private BloomFilter(BitSet bitSet, int numHashFunctions) {
Review Comment:
check `bitSet` not null, and check `numHashFunctions` > 0
##########
flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/BloomFilterTest.java:
##########
@@ -168,4 +173,161 @@ public void testHashcodeInput() {
assertTrue(bloomFilter.testHash(val4));
assertTrue(bloomFilter.testHash(val5));
}
+
+ @Test
+ public void testBloomFilterSerDe() {
Review Comment:
Can we migrate this test to junit5?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -143,4 +176,63 @@ public String toString() {
output.append(bitSet);
return output.toString();
}
+
+ public static byte[] toBytes(BloomFilter filter) {
Review Comment:
Add doc, and indicate that currently only supports heap.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -143,4 +176,63 @@ public String toString() {
output.append(bitSet);
return output.toString();
}
+
+ public static byte[] toBytes(BloomFilter filter) {
+ byte[] data = filter.bitSet.getData();
+ int byteSize = data.length;
+ byte[] bytes = new byte[8 + byteSize];
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET, filter.numHashFunctions);
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET + 4, byteSize);
+ UNSAFE.copyMemory(
+ data, BYTE_ARRAY_BASE_OFFSET, bytes, BYTE_ARRAY_BASE_OFFSET + 8, byteSize);
+ return bytes;
+ }
+
+ public static BloomFilter fromBytes(byte[] bytes) {
+ int numHashFunctions = UNSAFE.getInt(bytes, BYTE_ARRAY_BASE_OFFSET);
+ int byteSize = UNSAFE.getInt(bytes, BYTE_ARRAY_BASE_OFFSET + 4);
+ byte[] data = new byte[byteSize];
+ UNSAFE.copyMemory(
+ bytes, BYTE_ARRAY_BASE_OFFSET + 8, data, BYTE_ARRAY_BASE_OFFSET, byteSize);
+
+ BitSet bitSet = new BitSet(byteSize);
+ bitSet.setMemorySegment(MemorySegmentFactory.wrap(data), 0);
+ return new BloomFilter(bitSet, numHashFunctions);
+ }
+
+ public static byte[] mergeBloomFilterBytes(byte[] bf1Bytes, byte[] bf2Bytes) {
+ return mergeBloomFilterBytes(bf1Bytes, 0, bf1Bytes.length, bf2Bytes, 0, bf2Bytes.length);
+ }
+
+ /** Merge the bf2 bytes to bf1 */
+ public static byte[] mergeBloomFilterBytes(
Review Comment:
can be private
##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/BloomFilter.java:
##########
@@ -143,4 +176,63 @@ public String toString() {
output.append(bitSet);
return output.toString();
}
+
+ public static byte[] toBytes(BloomFilter filter) {
+ byte[] data = filter.bitSet.getData();
+ int byteSize = data.length;
+ byte[] bytes = new byte[8 + byteSize];
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET, filter.numHashFunctions);
+ UNSAFE.putInt(bytes, BYTE_ARRAY_BASE_OFFSET + 4, byteSize);
+ UNSAFE.copyMemory(
+ data, BYTE_ARRAY_BASE_OFFSET, bytes, BYTE_ARRAY_BASE_OFFSET + 8, byteSize);
+ return bytes;
+ }
+
+ public static BloomFilter fromBytes(byte[] bytes) {
+ int numHashFunctions = UNSAFE.getInt(bytes, BYTE_ARRAY_BASE_OFFSET);
+ int byteSize = UNSAFE.getInt(bytes, BYTE_ARRAY_BASE_OFFSET + 4);
+ byte[] data = new byte[byteSize];
+ UNSAFE.copyMemory(
+ bytes, BYTE_ARRAY_BASE_OFFSET + 8, data, BYTE_ARRAY_BASE_OFFSET, byteSize);
+
+ BitSet bitSet = new BitSet(byteSize);
+ bitSet.setMemorySegment(MemorySegmentFactory.wrap(data), 0);
+ return new BloomFilter(bitSet, numHashFunctions);
+ }
+
+ public static byte[] mergeBloomFilterBytes(byte[] bf1Bytes, byte[] bf2Bytes) {
+ return mergeBloomFilterBytes(bf1Bytes, 0, bf1Bytes.length, bf2Bytes, 0, bf2Bytes.length);
+ }
+
+ /** Merge the bf2 bytes to bf1 */
+ public static byte[] mergeBloomFilterBytes(
Review Comment:
Maybe named `mergeSerializedBloomFilters` ? And add doc to indicate that `bf1Bytes` will be changed after the merge.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/operators/util/BloomFilterTest.java:
##########
@@ -168,4 +173,161 @@ public void testHashcodeInput() {
assertTrue(bloomFilter.testHash(val4));
assertTrue(bloomFilter.testHash(val5));
}
+
+ @Test
+ public void testBloomFilterSerDe() {
Review Comment:
testBloomFilterSerializationAndDeserialization
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wanglijie95 closed pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "wanglijie95 (via GitHub)" <gi...@apache.org>.
wanglijie95 closed pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
URL: https://github.com/apache/flink/pull/22908
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on pull request #22908: [FLINK-32489][runtime] Support serialize and merge BloomFilter for runtime filter
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on PR #22908:
URL: https://github.com/apache/flink/pull/22908#issuecomment-1613503298
cc @wanglijie95
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org