You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/03/12 10:13:50 UTC
[flink] branch master updated: [FLINK-11856][table-runtime-blink]
Introduce BinaryHashTable to batch table runtime. (#7949)
This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5968206 [FLINK-11856][table-runtime-blink] Introduce BinaryHashTable to batch table runtime. (#7949)
5968206 is described below
commit 5968206f843a58e10e3db58cde3407bb3608d085
Author: Kurt Young <yk...@gmail.com>
AuthorDate: Tue Mar 12 18:13:37 2019 +0800
[FLINK-11856][table-runtime-blink] Introduce BinaryHashTable to batch table runtime. (#7949)
---
.../apache/flink/core/memory/MemorySegment.java | 8 +
.../runtime/memory/AbstractPagedOutputView.java | 31 +
.../apache/flink/table/api/TableConfigOptions.java | 13 +-
.../flink/table/dataformat/util/BinaryRowUtil.java | 59 ++
.../JoinCondition.java} | 28 +-
.../Projection.java} | 25 +-
.../runtime/hashtable/BaseHybridHashTable.java | 561 +++++++++++++
.../runtime/hashtable/BinaryHashBucketArea.java | 624 ++++++++++++++
.../runtime/hashtable/BinaryHashPartition.java | 608 ++++++++++++++
.../table/runtime/hashtable/BinaryHashTable.java | 666 +++++++++++++++
.../BinaryRowChannelInputViewIterator.java | 75 ++
.../table/runtime/hashtable/BuildSideIterator.java | 233 +++++
.../runtime/hashtable/HashTableBloomFilter.java | 88 ++
.../runtime/hashtable/LookupBucketIterator.java | 140 ++++
.../table/runtime/hashtable/ProbeIterator.java | 70 ++
.../runtime/hashtable/WrappedRowIterator.java} | 44 +-
.../flink/table/runtime/join/HashJoinType.java | 62 ++
.../table/runtime/join/NullAwareJoinHelper.java | 51 ++
.../table/runtime/sort/BinaryIndexedSortable.java | 1 -
.../runtime/sort/BinaryKVInMemorySortBuffer.java | 3 +-
.../table/runtime/sort/ListMemorySegmentPool.java | 1 -
.../table/runtime/util/MemorySegmentPool.java | 23 +-
.../flink/table/runtime/util/RowIterator.java | 46 +
.../table/typeutils/AbstractRowSerializer.java | 75 +-
.../flink/table/typeutils/BaseRowSerializer.java | 94 ++-
.../flink/table/typeutils/BinaryRowSerializer.java | 105 ++-
.../runtime/hashtable/BinaryHashTableTest.java | 933 +++++++++++++++++++++
.../table/runtime/sort/TestMemorySegmentPool.java | 4 -
.../runtime/util/UniformBinaryRowGenerator.java | 94 +++
29 files changed, 4573 insertions(+), 192 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
index e747324..614696c 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
@@ -1426,4 +1426,12 @@ public abstract class MemorySegment {
return true;
}
+
+ /**
+ * Get the heap byte array object.
+ * @return Return non-null if the memory is on the heap, and return null if the memory if off the heap.
+ */
+ public byte[] getHeapMemory() {
+ return heapMemory;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/AbstractPagedOutputView.java b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/AbstractPagedOutputView.java
index fbb6d5e..7451d54 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/memory/AbstractPagedOutputView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/memory/AbstractPagedOutputView.java
@@ -415,4 +415,35 @@ public abstract class AbstractPagedOutputView implements DataOutputView {
advance();
}
}
+
+ public void write(MemorySegment segment, int off, int len) throws IOException {
+ int remaining = this.segmentSize - this.positionInSegment;
+ if (remaining >= len) {
+ segment.copyTo(off, currentSegment, positionInSegment, len);
+ this.positionInSegment += len;
+ } else {
+
+ if (remaining == 0) {
+ advance();
+ remaining = this.segmentSize - this.positionInSegment;
+ }
+
+ while (true) {
+ int toPut = Math.min(remaining, len);
+ segment.copyTo(off, currentSegment, positionInSegment, toPut);
+ off += toPut;
+ len -= toPut;
+
+ if (len > 0) {
+ this.positionInSegment = this.segmentSize;
+ advance();
+ remaining = this.segmentSize - this.positionInSegment;
+ }
+ else {
+ this.positionInSegment += toPut;
+ break;
+ }
+ }
+ }
+ }
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/api/TableConfigOptions.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/api/TableConfigOptions.java
index b4afcc1..f9987d1 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/api/TableConfigOptions.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/api/TableConfigOptions.java
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -34,8 +34,9 @@ public class TableConfigOptions {
public static final ConfigOption<Integer> SQL_EXEC_SORT_FILE_HANDLES_MAX_NUM =
key("sql.exec.sort.file-handles.max.num")
.defaultValue(128)
- .withDescription("Sort merge's maximum number of roads, too many roads, may cause too many files to be" +
- " read at the same time, resulting in excessive use of memory.");
+ .withDescription("Sort merge's maximum number of roads, too many roads, " +
+ "may cause too many files to be read at the same time, resulting in " +
+ "excessive use of memory.");
public static final ConfigOption<Boolean> SQL_EXEC_SORT_ASYNC_MERGE_ENABLED =
key("sql.exec.sort.async-merge.enabled")
@@ -49,12 +50,14 @@ public class TableConfigOptions {
public static final ConfigOption<Boolean> SQL_EXEC_SPILL_COMPRESSION_ENABLED =
key("sql.exec.spill.compression.enabled")
.defaultValue(true)
- .withDescription("Whether to compress spilled data. (Now include sort and hash agg and hash join)");
+ .withDescription("Whether to compress spilled data. " +
+ "(Now include sort and hash agg and hash join)");
public static final ConfigOption<String> SQL_EXEC_SPILL_COMPRESSION_CODEC =
key("sql.exec.spill.compression.codec")
.defaultValue("lz4")
- .withDescription("Use that compression codec to compress spilled file. Now we support lz4, gzip, bzip2.");
+ .withDescription("Use that compression codec to compress spilled file. " +
+ "Now we support lz4.");
public static final ConfigOption<Integer> SQL_EXEC_SPILL_COMPRESSION_BLOCK_SIZE =
key("sql.exec.spill.compression.block-size")
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/util/BinaryRowUtil.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/util/BinaryRowUtil.java
new file mode 100644
index 0000000..71dc6c2
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/util/BinaryRowUtil.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.dataformat.util;
+
+import org.apache.flink.core.memory.MemoryUtils;
+
+/**
+ * Util for binary row. Many of the methods in this class are used in code generation.
+ * So ignore IDE warnings.
+ */
+public class BinaryRowUtil {
+
+ public static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
+ public static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
+
+ public static boolean byteArrayEquals(byte[] left, byte[] right, int length) {
+ return byteArrayEquals(
+ left, BYTE_ARRAY_BASE_OFFSET, right, BYTE_ARRAY_BASE_OFFSET, length);
+ }
+
+ public static boolean byteArrayEquals(
+ Object left, long leftOffset, Object right, long rightOffset, int length) {
+ int i = 0;
+
+ while (i <= length - 8) {
+ if (UNSAFE.getLong(left, leftOffset + i) !=
+ UNSAFE.getLong(right, rightOffset + i)) {
+ return false;
+ }
+ i += 8;
+ }
+
+ while (i < length) {
+ if (UNSAFE.getByte(left, leftOffset + i) !=
+ UNSAFE.getByte(right, rightOffset + i)) {
+ return false;
+ }
+ i += 1;
+ }
+ return true;
+ }
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/JoinCondition.java
similarity index 60%
copy from flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
copy to flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/JoinCondition.java
index 94713e5..b047b0f 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/JoinCondition.java
@@ -16,30 +16,18 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.util;
+package org.apache.flink.table.generated;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentSource;
-
-import java.util.Arrays;
-import java.util.List;
+import org.apache.flink.table.dataformat.BaseRow;
/**
- * MemorySegment pool.
+ * Interface for code generated condition function for [[org.apache.calcite.rel.core.Join]].
*/
-public interface MemorySegmentPool extends MemorySegmentSource {
-
- int pageSize();
-
- void returnAll(List<MemorySegment> memory);
-
- default void returnAll(MemorySegment[] memory) {
- returnAll(Arrays.asList(memory));
- }
+public interface JoinCondition {
- void clear();
+ /**
+ * @return true if the join condition stays true for the joined row (in1, in2)
+ */
+ boolean apply(BaseRow in1, BaseRow in2);
- default int remainBuffers() {
- throw new UnsupportedOperationException();
- }
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/Projection.java
similarity index 60%
copy from flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
copy to flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/Projection.java
index 94713e5..3d7bdf1 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/Projection.java
@@ -16,30 +16,15 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.util;
+package org.apache.flink.table.generated;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentSource;
-
-import java.util.Arrays;
-import java.util.List;
+import org.apache.flink.table.dataformat.BaseRow;
/**
- * MemorySegment pool.
+ * Interface for code generated projection, which will map a BaseRow to another one.
*/
-public interface MemorySegmentPool extends MemorySegmentSource {
-
- int pageSize();
-
- void returnAll(List<MemorySegment> memory);
-
- default void returnAll(MemorySegment[] memory) {
- returnAll(Arrays.asList(memory));
- }
+public interface Projection<IN extends BaseRow, OUT extends BaseRow> {
- void clear();
+ OUT apply(IN row);
- default int remainBuffers() {
- throw new UnsupportedOperationException();
- }
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BaseHybridHashTable.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BaseHybridHashTable.java
new file mode 100644
index 0000000..d06fce6
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BaseHybridHashTable.java
@@ -0,0 +1,561 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.AbstractChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.HeaderlessChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.api.TableConfigOptions;
+import org.apache.flink.table.runtime.compression.BlockCompressionFactory;
+import org.apache.flink.table.runtime.util.FileChannelUtil;
+import org.apache.flink.table.runtime.util.MemorySegmentPool;
+import org.apache.flink.util.MathUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.flink.core.memory.MemorySegmentFactory.allocateUnpooledSegment;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Base table for {@link LongHybridHashTable} and {@link BinaryHashTable}.
+ */
+public abstract class BaseHybridHashTable implements MemorySegmentPool {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(BaseHybridHashTable.class);
+
+ /**
+ * The maximum number of recursive partitionings that the join does before giving up.
+ */
+ protected static final int MAX_RECURSION_DEPTH = 3;
+
+ /**
+ * The maximum number of partitions, which defines the spilling granularity. Each recursion,
+ * the data is divided maximally into that many partitions, which are processed in one chuck.
+ */
+ protected static final int MAX_NUM_PARTITIONS = Byte.MAX_VALUE;
+
+ /**
+ * The minimum number of memory segments the hash join needs to be supplied with in order to
+ * work.
+ */
+ private static final int MIN_NUM_MEMORY_SEGMENTS = 33;
+ protected final int initPartitionFanOut;
+
+ /**
+ * The owner to associate with the memory segment.
+ */
+ private Object owner;
+ private final int avgRecordLen;
+ protected final long buildRowCount;
+
+ /**
+ * The total reserved number of memory segments available to the hash join.
+ */
+ protected final int reservedNumBuffers;
+ /**
+ * The total max number of memory segments available to the hash join.
+ */
+ private final int maxNumBuffers;
+
+ /**
+ * record number of the allocated segments from the floating pool.
+ */
+ protected int allocatedFloatingNum;
+ private final int perRequestNumBuffers;
+ private final MemoryManager memManager;
+
+ /**
+ * The free memory segments currently available to the hash join.
+ */
+ public final ArrayList<MemorySegment> availableMemory;
+
+ /**
+ * The I/O manager used to instantiate writers for the spilled partitions.
+ */
+ protected final IOManager ioManager;
+
+ /**
+ * The size of the segments used by the hash join buckets. All segments must be of equal size to
+ * ease offset computations.
+ */
+ protected final int segmentSize;
+
+ /**
+ * The queue of buffers that can be used for write-behind. Any buffer that is written
+ * asynchronously to disk is returned through this queue. hence
+ */
+ protected final LinkedBlockingQueue<MemorySegment> buildSpillReturnBuffers;
+
+ public final int segmentSizeBits;
+
+ public final int segmentSizeMask;
+
+ /**
+ * Flag indicating that the closing logic has been invoked.
+ */
+ protected AtomicBoolean closed = new AtomicBoolean();
+
+ /**
+ * Try to make the buildSide rows distinct.
+ */
+ public final boolean tryDistinctBuildRow;
+
+ /**
+ * The recursion depth of the partition that is currently processed. The initial table
+ * has a recursion depth of 0. Partitions spilled from a table that is built for a partition
+ * with recursion depth <i>n</i> have a recursion depth of <i>n+1</i>.
+ */
+ protected int currentRecursionDepth;
+
+ /**
+ * The number of buffers in the build spill return buffer queue that are actually not write behind buffers,
+ * but regular buffers that only have not yet returned. This is part of an optimization that the
+ * spilling code needs not wait until the partition is completely spilled before proceeding.
+ */
+ protected int buildSpillRetBufferNumbers;
+
+ /**
+ * The reader for the spilled-file of the build partition that is currently read.
+ */
+ protected HeaderlessChannelReaderInputView currentSpilledBuildSide;
+ /**
+ * The reader for the spilled-file of the probe partition that is currently read.
+ */
+ protected AbstractChannelReaderInputView currentSpilledProbeSide;
+
+ /**
+ * The channel enumerator that is used while processing the current partition to create
+ * channels for the spill partitions it requires.
+ */
+ protected FileIOChannel.Enumerator currentEnumerator;
+
+ protected final boolean compressionEnable;
+ protected final BlockCompressionFactory compressionCodecFactory;
+ protected final int compressionBlockSize;
+
+ protected transient long numSpillFiles;
+ protected transient long spillInBytes;
+
+ public BaseHybridHashTable(
+ Configuration conf,
+ Object owner,
+ MemoryManager memManager,
+ long reservedMemorySize,
+ long preferredMemorySize,
+ long perRequestMemorySize,
+ IOManager ioManager,
+ int avgRecordLen,
+ long buildRowCount,
+ boolean tryDistinctBuildRow) {
+
+ //TODO: read compression config from configuration
+ this.compressionEnable = conf.getBoolean(TableConfigOptions.SQL_EXEC_SPILL_COMPRESSION_ENABLED);
+ this.compressionCodecFactory = this.compressionEnable
+ ? BlockCompressionFactory.createBlockCompressionFactory(
+ conf.getString(TableConfigOptions.SQL_EXEC_SPILL_COMPRESSION_CODEC))
+ : null;
+ this.compressionBlockSize = conf.getInteger(TableConfigOptions.SQL_EXEC_SPILL_COMPRESSION_BLOCK_SIZE);
+
+ this.owner = owner;
+ this.avgRecordLen = avgRecordLen;
+ this.buildRowCount = buildRowCount;
+ this.tryDistinctBuildRow = tryDistinctBuildRow;
+ this.reservedNumBuffers = (int) (reservedMemorySize / memManager.getPageSize());
+ // some sanity checks first
+ checkArgument(reservedNumBuffers >= MIN_NUM_MEMORY_SEGMENTS);
+ this.maxNumBuffers = (int) (preferredMemorySize / memManager.getPageSize());
+ this.perRequestNumBuffers = (int) (perRequestMemorySize / memManager.getPageSize());
+ this.availableMemory = new ArrayList<>(this.reservedNumBuffers);
+ try {
+ List<MemorySegment> allocates = memManager.allocatePages(owner, this.reservedNumBuffers);
+ this.availableMemory.addAll(allocates);
+ allocates.clear();
+ } catch (MemoryAllocationException e) {
+ LOG.error("Out of memory", e);
+ throw new RuntimeException(e);
+ }
+ this.memManager = memManager;
+ this.ioManager = ioManager;
+
+ this.segmentSize = memManager.getPageSize();
+
+ checkArgument(MathUtils.isPowerOf2(segmentSize));
+
+ // take away the write behind buffers
+ this.buildSpillReturnBuffers = new LinkedBlockingQueue<>();
+
+ this.segmentSizeBits = MathUtils.log2strict(segmentSize);
+ this.segmentSizeMask = segmentSize - 1;
+
+ // open builds the initial table by consuming the build-side input
+ this.currentRecursionDepth = 0;
+
+ // create the partitions
+ this.initPartitionFanOut = Math.min(getPartitioningFanOutNoEstimates(), maxNumPartition());
+
+ this.closed.set(false);
+
+ LOG.info(String.format("Initialize hash table with %d memory segments, each size [%d], the reserved memory %d" +
+ " MB, the max memory %d MB, per allocate {} segments from floating memory pool.",
+ reservedNumBuffers, segmentSize, (long) reservedNumBuffers * segmentSize / 1024 / 1024,
+ (long) maxNumBuffers * segmentSize / 1024 / 1024), perRequestNumBuffers);
+ }
+
+ /**
+ * Bucket area need at-least one and data need at-least one.
+ * In the initialization phase, we can use (totalNumBuffers - numWriteBehindBuffers) Segments.
+ * However, in the buildTableFromSpilledPartition phase, only (totalNumBuffers - numWriteBehindBuffers - 2)
+ * can be used because two Buffers are needed to read the data.
+ */
+ protected int maxNumPartition() {
+ return (availableMemory.size() + buildSpillRetBufferNumbers) / 2;
+ }
+
+ /**
+ * Gets the number of partitions to be used for an initial hash-table.
+ */
+ private int getPartitioningFanOutNoEstimates() {
+ return Math.max(11, findSmallerPrime((int) Math.min(buildRowCount * avgRecordLen / (10 * segmentSize),
+ MAX_NUM_PARTITIONS)));
+ }
+
+ /**
+ * Let prime number be the numBuckets, to avoid partition hash and bucket hash congruences.
+ */
+ private static int findSmallerPrime(int num) {
+ for (; num > 1; num--) {
+ if (isPrimeNumber(num)) {
+ return num;
+ }
+ }
+ return num;
+ }
+
+ private static boolean isPrimeNumber(int num){
+ if (num == 2) {
+ return true;
+ }
+ if (num < 2 || num % 2 == 0) {
+ return false;
+ }
+ for (int i = 3; i <= Math.sqrt(num); i += 2){
+ if (num % i == 0) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Gets the next buffer to be used with the hash-table, either for an in-memory partition, or
+ * for the table buckets. This method returns <tt>null</tt>, if no more buffer is available.
+ * Spilling a partition may free new buffers then.
+ *
+ * @return The next buffer to be used by the hash-table, or null, if no buffer remains.
+ */
+ public MemorySegment getNextBuffer() {
+ // check if the list directly offers memory
+ int s = this.availableMemory.size();
+ if (s > 0) {
+ return this.availableMemory.remove(s - 1);
+ }
+
+ // check if there are write behind buffers that actually are to be used for the hash table
+ if (this.buildSpillRetBufferNumbers > 0) {
+ // grab at least one, no matter what
+ MemorySegment toReturn;
+ try {
+ toReturn = this.buildSpillReturnBuffers.take();
+ } catch (InterruptedException iex) {
+ throw new RuntimeException("Hybrid Hash Join was interrupted while taking a buffer.");
+ }
+ this.buildSpillRetBufferNumbers--;
+
+ // grab as many more buffers as are available directly
+ MemorySegment currBuff;
+ while (this.buildSpillRetBufferNumbers > 0 && (currBuff = this.buildSpillReturnBuffers.poll()) != null) {
+ this.availableMemory.add(currBuff);
+ this.buildSpillRetBufferNumbers--;
+ }
+ return toReturn;
+ } else {
+ if (reservedNumBuffers + allocatedFloatingNum >= maxNumBuffers) {
+ //no more memory.
+ return null;
+ } else {
+ int requestNum = Math.min(perRequestNumBuffers, maxNumBuffers - reservedNumBuffers -
+ allocatedFloatingNum);
+ //apply for much more memory.
+ try {
+ List<MemorySegment> allocates = memManager.allocatePages(owner, requestNum);
+ this.availableMemory.addAll(allocates);
+ allocatedFloatingNum += allocates.size();
+ allocates.clear();
+ LOG.info("{} allocate {} floating segments successfully!", owner, requestNum);
+ } catch (MemoryAllocationException e) {
+ LOG.warn("BinaryHashMap can't allocate {} floating pages, and now used {} pages",
+ requestNum, reservedNumBuffers + allocatedFloatingNum, e);
+ //can't allocate much more memory.
+ return null;
+ }
+ if (this.availableMemory.size() > 0) {
+ return this.availableMemory.remove(this.availableMemory.size() - 1);
+ } else {
+ return null;
+ }
+ }
+ }
+ }
+
+ /**
+ * Bulk memory acquisition.
+ * NOTE: Failure to get memory will throw an exception.
+ */
+ public MemorySegment[] getNextBuffers(int bufferSize) {
+ MemorySegment[] memorySegments = new MemorySegment[bufferSize];
+ for (int i = 0; i < bufferSize; i++) {
+ MemorySegment nextBuffer = getNextBuffer();
+ if (nextBuffer == null) {
+ throw new RuntimeException("No enough buffers!");
+ }
+ memorySegments[i] = nextBuffer;
+ }
+ return memorySegments;
+ }
+
+ protected MemorySegment getNotNullNextBuffer() {
+ MemorySegment buffer = getNextBuffer();
+ if (buffer == null) {
+ throw new RuntimeException("Bug in HybridHashJoin: No memory became available.");
+ }
+ return buffer;
+ }
+
+ /**
+ * This is the method called by the partitions to request memory to serialize records.
+ * It automatically spills partitions, if memory runs out.
+ *
+ * @return The next available memory segment.
+ */
+ @Override
+ public MemorySegment nextSegment() {
+ final MemorySegment seg = getNextBuffer();
+ if (seg != null) {
+ return seg;
+ } else {
+ try {
+ spillPartition();
+ } catch (IOException ioex) {
+ throw new RuntimeException("Error spilling Hash Join Partition" + (ioex.getMessage() == null ?
+ "." : ": " + ioex.getMessage()), ioex);
+ }
+
+ MemorySegment fromSpill = getNextBuffer();
+ if (fromSpill == null) {
+ throw new RuntimeException("BUG in Hybrid Hash Join: Spilling did not free a buffer.");
+ } else {
+ return fromSpill;
+ }
+ }
+ }
+
+ @Override
+ public int pageSize() {
+ return segmentSize;
+ }
+
+ @Override
+ public void returnAll(List<MemorySegment> memory) {
+ for (MemorySegment segment : memory) {
+ if (segment != null) {
+ availableMemory.add(segment);
+ }
+ }
+ }
+
+ protected abstract int spillPartition() throws IOException;
+
+ /**
+ * This method makes sure that at least a certain number of memory segments is in the list of
+ * free segments.
+ * Free memory can be in the list of free segments, or in the return-queue where segments used
+ * to write behind are
+ * put. The number of segments that are in that return-queue, but are actually reclaimable is
+ * tracked. This method
+ * makes sure at least a certain number of buffers is reclaimed.
+ *
+ * @param minRequiredAvailable The minimum number of buffers that needs to be reclaimed.
+ */
+ public void ensureNumBuffersReturned(final int minRequiredAvailable) {
+ if (minRequiredAvailable > this.availableMemory.size() + this.buildSpillRetBufferNumbers) {
+ throw new IllegalArgumentException("More buffers requested available than totally available.");
+ }
+
+ try {
+ while (this.availableMemory.size() < minRequiredAvailable) {
+ this.availableMemory.add(this.buildSpillReturnBuffers.take());
+ this.buildSpillRetBufferNumbers--;
+ }
+ } catch (InterruptedException iex) {
+ throw new RuntimeException("Hash Join was interrupted.");
+ }
+ }
+
+ /**
+ * Closes the hash table. This effectively releases all internal structures and closes all
+ * open files and removes them. The call to this method is valid both as a cleanup after the
+ * complete inputs were properly processed, and as an cancellation call, which cleans up
+ * all resources that are currently held by the hash join.
+ */
+ public void close() {
+ // make sure that we close only once
+ if (!this.closed.compareAndSet(false, true)) {
+ return;
+ }
+
+ // clear the current probe side channel, if there is one
+ if (this.currentSpilledProbeSide != null) {
+ try {
+ this.currentSpilledProbeSide.getChannel().closeAndDelete();
+ } catch (Throwable t) {
+ LOG.warn("Could not close and delete the temp file for the current spilled partition probe side.", t);
+ }
+ }
+
+ // clear the memory in the partitions
+ clearPartitions();
+
+ // return the write-behind buffers
+ for (int i = 0; i < this.buildSpillRetBufferNumbers; i++) {
+ try {
+ this.availableMemory.add(this.buildSpillReturnBuffers.take());
+ } catch (InterruptedException iex) {
+ throw new RuntimeException("Hashtable closing was interrupted");
+ }
+ }
+ this.buildSpillRetBufferNumbers = 0;
+ }
+
+ protected abstract void clearPartitions();
+
+ public void free() {
+ if (this.closed.get()) {
+ memManager.release(availableMemory);
+ allocatedFloatingNum = 0;
+ } else {
+ throw new IllegalStateException("Cannot release memory until BinaryHashTable is closed!");
+ }
+ }
+
+ /**
+ * Free the memory not used.
+ */
+ public void freeCurrent() {
+ int beforeReleaseNum = availableMemory.size();
+ memManager.release(availableMemory);
+ allocatedFloatingNum -= (beforeReleaseNum - availableMemory.size());
+ }
+
+ @VisibleForTesting
+ public List<MemorySegment> getFreedMemory() {
+ return this.availableMemory;
+ }
+
+ public void free(MemorySegment segment) {
+ this.availableMemory.add(segment);
+ }
+
+ public int remainBuffers() {
+ return availableMemory.size() + buildSpillRetBufferNumbers;
+ }
+
+ public long getUsedMemoryInBytes() {
+ return (reservedNumBuffers + allocatedFloatingNum - availableMemory.size()) *
+ ((long) memManager.getPageSize());
+ }
+
+ public long getNumSpillFiles() {
+ return numSpillFiles;
+ }
+
+ public long getSpillInBytes() {
+ return spillInBytes;
+ }
+
+ /**
+ * Give up to one-sixth of the memory of the bucket area.
+ */
+ public int maxInitBufferOfBucketArea(int partitions) {
+ return Math.max(1, ((reservedNumBuffers - 2) / 6) / partitions);
+ }
+
+ protected List<MemorySegment> readAllBuffers(FileIOChannel.ID id, int blockCount) throws IOException {
+ // we are guaranteed to stay in memory
+ ensureNumBuffersReturned(blockCount);
+
+ LinkedBlockingQueue<MemorySegment> retSegments = new LinkedBlockingQueue<>();
+ BlockChannelReader<MemorySegment> reader = FileChannelUtil.createBlockChannelReader(
+ ioManager, id, retSegments,
+ compressionEnable, compressionCodecFactory, compressionBlockSize, segmentSize);
+ for (int i = 0; i < blockCount; i++) {
+ reader.readBlock(availableMemory.remove(availableMemory.size() - 1));
+ }
+ reader.closeAndDelete();
+
+ final List<MemorySegment> buffers = new ArrayList<>();
+ retSegments.drainTo(buffers);
+ return buffers;
+ }
+
+ protected HeaderlessChannelReaderInputView createInputView(FileIOChannel.ID id, int blockCount, int lastSegmentLimit) throws IOException {
+ BlockChannelReader<MemorySegment> inReader = FileChannelUtil.createBlockChannelReader(
+ ioManager, id, new LinkedBlockingQueue<>(),
+ compressionEnable, compressionCodecFactory, compressionBlockSize, segmentSize);
+ return new HeaderlessChannelReaderInputView(inReader,
+ Arrays.asList(allocateUnpooledSegment(segmentSize), allocateUnpooledSegment(segmentSize)),
+ blockCount, lastSegmentLimit, false);
+
+ }
+
+ /**
+ * The level parameter is needed so that we can have different hash functions when we
+ * recursively apply the partitioning, so that the working set eventually fits into memory.
+ */
+ public static int hash(int hashCode, int level) {
+ final int rotation = level * 11;
+ int code = Integer.rotateLeft(hashCode, rotation);
+ return code >= 0 ? code : -(code + 1);
+ }
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashBucketArea.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashBucketArea.java
new file mode 100644
index 0000000..9665b30
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashBucketArea.java
@@ -0,0 +1,624 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.RandomAccessInputView;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.util.MathUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Bucket area for hash table.
+ *
+ * <p>The layout of the buckets inside a memory segment is as follows:</p>
+ * <pre>
+ * +----------------------------- Bucket x ----------------------------
+ * |element count (2 bytes) | probedFlags (2 bytes) | next-bucket-in-chain-pointer (4 bytes) |
+ * |
+ * |hashCode 1 (4 bytes) | hashCode 2 (4 bytes) | hashCode 3 (4 bytes) |
+ * | ... hashCode n-1 (4 bytes) | hashCode n (4 bytes)
+ * |
+ * |pointer 1 (4 bytes) | pointer 2 (4 bytes) | pointer 3 (4 bytes) |
+ * | ... pointer n-1 (4 bytes) | pointer n (4 bytes)
+ * |
+ * +---------------------------- Bucket x + 1--------------------------
+ * | ...
+ * |
+ * </pre>
+ */
+public class BinaryHashBucketArea {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BinaryHashBucketArea.class);
+
+ /**
+ * Log 2 of bucket size.
+ */
+ static final int BUCKET_SIZE_BITS = 7;
+
+ /**
+ * 128, bucket size of bytes.
+ */
+ static final int BUCKET_SIZE = 0x1 << BUCKET_SIZE_BITS;
+
+ /**
+ * The length of the hash code stored in the bucket.
+ */
+ static final int HASH_CODE_LEN = 4;
+
+ /**
+ * The length of a pointer from a hash bucket to the record in the buffers.
+ */
+ static final int POINTER_LEN = 4;
+
+ /**
+ * The number of bytes that the entry in the hash structure occupies, in bytes.
+ * It corresponds to a 4 byte hash value and an 4 byte pointer.
+ */
+ public static final int RECORD_BYTES = HASH_CODE_LEN + POINTER_LEN;
+
+ /**
+ * Offset of the field in the bucket header indicating the bucket's element count.
+ */
+ static final int HEADER_COUNT_OFFSET = 0;
+
+ /**
+ * Offset of the field in the bucket header that holds the probed bit set.
+ */
+ static final int PROBED_FLAG_OFFSET = 2;
+
+ /**
+ * Offset of the field in the bucket header that holds the forward pointer to its
+ * first overflow bucket.
+ */
+ static final int HEADER_FORWARD_OFFSET = 4;
+
+ /**
+ * Total length for bucket header.
+ */
+ static final int BUCKET_HEADER_LENGTH = 8;
+
+ /**
+ * The maximum number of elements that can be loaded in a bucket.
+ */
+ static final int NUM_ENTRIES_PER_BUCKET = (BUCKET_SIZE - BUCKET_HEADER_LENGTH) / RECORD_BYTES;
+
+ /**
+ * Offset of record pointer.
+ */
+ static final int BUCKET_POINTER_START_OFFSET = BUCKET_HEADER_LENGTH + HASH_CODE_LEN * NUM_ENTRIES_PER_BUCKET;
+
+ /**
+ * Constant for the forward pointer, indicating that the pointer is not set.
+ */
+ static final int BUCKET_FORWARD_POINTER_NOT_SET = 0xFFFFFFFF;
+
+ /**
+ * Constant for the bucket header to init. (count: 0, probedFlag: 0, forwardPointerNotSet: ~0x0)
+ */
+ private static final long BUCKET_HEADER_INIT;
+
+ static {
+ if (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN) {
+ BUCKET_HEADER_INIT = 0xFFFFFFFF00000000L;
+ } else {
+ BUCKET_HEADER_INIT = 0x00000000FFFFFFFFL;
+ }
+ }
+
+ /**
+ * The load factor used when none specified in constructor.
+ */
+ private static final double DEFAULT_LOAD_FACTOR = 0.75;
+
+ final BinaryHashTable table;
+ private final double estimatedRowCount;
+ private final double loadFactor;
+ BinaryHashPartition partition;
+ private int size;
+
+ MemorySegment[] buckets;
+ int numBuckets;
+ private int numBucketsMask;
+ // segments in which overflow buckets from the table structure are stored
+ MemorySegment[] overflowSegments;
+ int numOverflowSegments; // the number of actual segments in the overflowSegments array
+ private int nextOverflowBucket; // the next free bucket in the current overflow segment
+ private int threshold;
+
+ private boolean inReHash = false;
+
+ BinaryHashBucketArea(BinaryHashTable table, double estimatedRowCount, int maxSegs) {
+ this(table, estimatedRowCount, maxSegs, DEFAULT_LOAD_FACTOR);
+ }
+
+ private BinaryHashBucketArea(BinaryHashTable table, double estimatedRowCount, int maxSegs, double loadFactor) {
+ this.table = table;
+ this.estimatedRowCount = estimatedRowCount;
+ this.loadFactor = loadFactor;
+ this.size = 0;
+
+ int minNumBuckets = (int) Math.ceil((estimatedRowCount / loadFactor / NUM_ENTRIES_PER_BUCKET));
+ int bucketNumSegs = Math.max(1, Math.min(maxSegs, (minNumBuckets >>> table.bucketsPerSegmentBits) +
+ ((minNumBuckets & table.bucketsPerSegmentMask) == 0 ? 0 : 1)));
+ int numBuckets = MathUtils.roundDownToPowerOf2(bucketNumSegs << table.bucketsPerSegmentBits);
+
+ int threshold = (int) (numBuckets * NUM_ENTRIES_PER_BUCKET * loadFactor);
+
+ MemorySegment[] buckets = new MemorySegment[bucketNumSegs];
+ table.ensureNumBuffersReturned(bucketNumSegs);
+
+ // go over all segments that are part of the table
+ for (int i = 0; i < bucketNumSegs; i++) {
+ final MemorySegment seg = table.getNextBuffer();
+ initMemorySegment(seg);
+ buckets[i] = seg;
+ }
+
+ setNewBuckets(buckets, numBuckets, threshold);
+ }
+
+ private void setNewBuckets(MemorySegment[] buckets, int numBuckets, int threshold) {
+ this.buckets = buckets;
+ checkArgument(MathUtils.isPowerOf2(numBuckets));
+ this.numBuckets = numBuckets;
+ this.numBucketsMask = numBuckets - 1;
+ this.overflowSegments = new MemorySegment[2];
+ this.numOverflowSegments = 0;
+ this.nextOverflowBucket = 0;
+ this.threshold = threshold;
+ }
+
+ public void setPartition(BinaryHashPartition partition) {
+ this.partition = partition;
+ }
+
+ private void resize(boolean spillingAllowed) throws IOException {
+ MemorySegment[] oldBuckets = this.buckets;
+ int oldNumBuckets = numBuckets;
+ MemorySegment[] oldOverflowSegments = overflowSegments;
+ int newNumSegs = oldBuckets.length * 2;
+ int newNumBuckets = MathUtils.roundDownToPowerOf2(newNumSegs << table.bucketsPerSegmentBits);
+ int newThreshold = (int) (newNumBuckets * NUM_ENTRIES_PER_BUCKET * loadFactor);
+
+ // We can't resize if not spillingAllowed and there are not enough buffers.
+ if (!spillingAllowed && newNumSegs > table.remainBuffers()) {
+ return;
+ }
+
+ // request new buckets.
+ MemorySegment[] newBuckets = new MemorySegment[newNumSegs];
+ for (int i = 0; i < newNumSegs; i++) {
+ MemorySegment seg = table.getNextBuffer();
+ if (seg == null) {
+ final int spilledPart = table.spillPartition();
+ if (spilledPart == partition.partitionNumber) {
+ // this bucket is no longer in-memory
+ // free new segments.
+ for (int j = 0; j < i; j++) {
+ table.free(newBuckets[j]);
+ }
+ return;
+ }
+ seg = table.getNextBuffer();
+ if (seg == null) {
+ throw new RuntimeException(
+ "Bug in HybridHashJoin: No memory became available after spilling a partition.");
+ }
+ }
+ initMemorySegment(seg);
+ newBuckets[i] = seg;
+ }
+
+ setNewBuckets(newBuckets, newNumBuckets, newThreshold);
+
+ reHash(oldBuckets, oldNumBuckets, oldOverflowSegments);
+ }
+
+ private void reHash(
+ MemorySegment[] oldBuckets,
+ int oldNumBuckets,
+ MemorySegment[] oldOverflowSegments) throws IOException {
+ long reHashStartTime = System.currentTimeMillis();
+ inReHash = true;
+ int scanCount = -1;
+ while (true) {
+ scanCount++;
+ if (scanCount >= oldNumBuckets) {
+ break;
+ }
+ // move to next bucket, update all the current bucket status with new bucket information.
+ final int bucketArrayPos = scanCount >> table.bucketsPerSegmentBits;
+ int bucketInSegOffset = (scanCount & table.bucketsPerSegmentMask) << BUCKET_SIZE_BITS;
+ MemorySegment bucketSeg = oldBuckets[bucketArrayPos];
+
+ int countInBucket = bucketSeg.getShort(bucketInSegOffset + HEADER_COUNT_OFFSET);
+ int numInBucket = 0;
+ while (countInBucket != 0) {
+ int hashCodeOffset = bucketInSegOffset + BUCKET_HEADER_LENGTH;
+ int pointerOffset = bucketInSegOffset + BUCKET_POINTER_START_OFFSET;
+ while (numInBucket < countInBucket) {
+ int hashCode = bucketSeg.getInt(hashCodeOffset);
+ int pointer = bucketSeg.getInt(pointerOffset);
+ if (!insertToBucket(hashCode, pointer, true, false)) {
+ buildBloomFilterAndFree(oldBuckets, oldNumBuckets, oldOverflowSegments);
+ return;
+ }
+ numInBucket++;
+ hashCodeOffset += HASH_CODE_LEN;
+ pointerOffset += POINTER_LEN;
+ }
+
+ // this segment is done. check if there is another chained bucket
+ int forwardPointer = bucketSeg.getInt(bucketInSegOffset + HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
+ break;
+ }
+
+ final int overflowSegIndex = forwardPointer >>> table.segmentSizeBits;
+ bucketSeg = oldOverflowSegments[overflowSegIndex];
+ bucketInSegOffset = forwardPointer & table.segmentSizeMask;
+ countInBucket = bucketSeg.getShort(bucketInSegOffset + HEADER_COUNT_OFFSET);
+ numInBucket = 0;
+ }
+ }
+
+ freeMemory(oldBuckets, oldOverflowSegments);
+ inReHash = false;
+ LOG.info("The rehash take {} ms for {} segments", (System.currentTimeMillis() - reHashStartTime), numBuckets);
+ }
+
+ private void freeMemory(MemorySegment[] buckets, MemorySegment[] overflowSegments) {
+ for (MemorySegment segment : buckets) {
+ table.free(segment);
+ }
+ for (MemorySegment segment : overflowSegments) {
+ if (segment != null) {
+ table.free(segment);
+ }
+ }
+ }
+
+ private void initMemorySegment(MemorySegment seg) {
+ // go over all buckets in the segment
+ for (int k = 0; k < table.bucketsPerSegment; k++) {
+ final int bucketOffset = k * BUCKET_SIZE;
+
+ // init count and probeFlag and forward pointer together.
+ seg.putLong(bucketOffset + HEADER_COUNT_OFFSET, BUCKET_HEADER_INIT);
+ }
+ }
+
+ private boolean insertToBucket(
+ MemorySegment bucket, int bucketInSegmentPos,
+ int hashCode, int pointer, boolean spillingAllowed, boolean sizeAddAndCheckResize) throws IOException {
+ final int count = bucket.getShort(bucketInSegmentPos + HEADER_COUNT_OFFSET);
+ if (count < NUM_ENTRIES_PER_BUCKET) {
+ // we are good in our current bucket, put the values
+ bucket.putShort(bucketInSegmentPos + HEADER_COUNT_OFFSET, (short) (count + 1)); // update count
+ bucket.putInt(bucketInSegmentPos + BUCKET_HEADER_LENGTH + (count * HASH_CODE_LEN), hashCode); // hash code
+ bucket.putInt(bucketInSegmentPos + BUCKET_POINTER_START_OFFSET + (count * POINTER_LEN), pointer); // pointer
+ } else {
+ // we need to go to the overflow buckets
+ final int originalForwardPointer = bucket.getInt(bucketInSegmentPos + HEADER_FORWARD_OFFSET);
+ final int forwardForNewBucket;
+
+ if (originalForwardPointer != BUCKET_FORWARD_POINTER_NOT_SET) {
+
+ // forward pointer set
+ final int overflowSegIndex = originalForwardPointer >>> table.segmentSizeBits;
+ final int segOffset = originalForwardPointer & table.segmentSizeMask;
+ final MemorySegment seg = overflowSegments[overflowSegIndex];
+
+ final short obCount = seg.getShort(segOffset + HEADER_COUNT_OFFSET);
+
+ // check if there is space in this overflow bucket
+ if (obCount < NUM_ENTRIES_PER_BUCKET) {
+ // space in this bucket and we are done
+ seg.putShort(segOffset + HEADER_COUNT_OFFSET, (short) (obCount + 1)); // update count
+ seg.putInt(segOffset + BUCKET_HEADER_LENGTH + (obCount * HASH_CODE_LEN), hashCode); // hash code
+ seg.putInt(segOffset + BUCKET_POINTER_START_OFFSET + (obCount * POINTER_LEN), pointer); // pointer
+ return true;
+ } else {
+ // no space here, we need a new bucket. this current overflow bucket will be the
+ // target of the new overflow bucket
+ forwardForNewBucket = originalForwardPointer;
+ }
+ } else {
+ // no overflow bucket yet, so we need a first one
+ forwardForNewBucket = BUCKET_FORWARD_POINTER_NOT_SET;
+ }
+
+ // we need a new overflow bucket
+ MemorySegment overflowSeg;
+ final int overflowBucketNum;
+ final int overflowBucketOffset;
+
+ // first, see if there is space for an overflow bucket remaining in the last overflow segment
+ if (nextOverflowBucket == 0) {
+ // no space left in last bucket, or no bucket yet, so create an overflow segment
+ overflowSeg = table.getNextBuffer();
+ if (overflowSeg == null) {
+ // no memory available to create overflow bucket. we need to spill a partition
+ if (!spillingAllowed) {
+ throw new IOException("Hashtable memory ran out in a non-spillable situation. " +
+ "This is probably related to wrong size calculations.");
+ }
+ final int spilledPart = table.spillPartition();
+ if (spilledPart == partition.partitionNumber) {
+ // this bucket is no longer in-memory
+ return false;
+ }
+ overflowSeg = table.getNextBuffer();
+ if (overflowSeg == null) {
+ throw new RuntimeException("Bug in HybridHashJoin: No memory became available after spilling a partition.");
+ }
+ }
+ overflowBucketOffset = 0;
+ overflowBucketNum = numOverflowSegments;
+
+ // add the new overflow segment
+ if (overflowSegments.length <= numOverflowSegments) {
+ MemorySegment[] newSegsArray = new MemorySegment[overflowSegments.length * 2];
+ System.arraycopy(overflowSegments, 0, newSegsArray, 0, overflowSegments.length);
+ overflowSegments = newSegsArray;
+ }
+ overflowSegments[numOverflowSegments] = overflowSeg;
+ numOverflowSegments++;
+ } else {
+ // there is space in the last overflow bucket
+ overflowBucketNum = numOverflowSegments - 1;
+ overflowSeg = overflowSegments[overflowBucketNum];
+ overflowBucketOffset = nextOverflowBucket << BUCKET_SIZE_BITS;
+ }
+
+ // next overflow bucket is one ahead. if the segment is full, the next will be at the beginning
+ // of a new segment
+ nextOverflowBucket = (nextOverflowBucket == table.bucketsPerSegmentMask ? 0 : nextOverflowBucket + 1);
+
+ // insert the new overflow bucket in the chain of buckets
+ // 1) set the old forward pointer
+ // 2) let the bucket in the main table point to this one
+ overflowSeg.putInt(overflowBucketOffset + HEADER_FORWARD_OFFSET, forwardForNewBucket);
+ final int pointerToNewBucket = (overflowBucketNum << table.segmentSizeBits) + overflowBucketOffset;
+ bucket.putInt(bucketInSegmentPos + HEADER_FORWARD_OFFSET, pointerToNewBucket);
+
+ // finally, insert the values into the overflow buckets
+ overflowSeg.putInt(overflowBucketOffset + BUCKET_HEADER_LENGTH, hashCode); // hash code
+ overflowSeg.putInt(overflowBucketOffset + BUCKET_POINTER_START_OFFSET, pointer); // pointer
+
+ // set the count to one
+ overflowSeg.putShort(overflowBucketOffset + HEADER_COUNT_OFFSET, (short) 1);
+
+ // initiate the probed bitset to 0.
+ overflowSeg.putShort(overflowBucketOffset + PROBED_FLAG_OFFSET, (short) 0);
+ }
+
+ if (sizeAddAndCheckResize && ++size > threshold) {
+ resize(spillingAllowed);
+ }
+ return true;
+ }
+
+ private int findBucket(int hashCode) {
+ return hashCode & this.numBucketsMask;
+ }
+
+ /**
+ * Insert into bucket by hashCode and pointer.
+ * @return return false when spill own partition.
+ */
+ boolean insertToBucket(int hashCode, int pointer, boolean spillingAllowed, boolean sizeAddAndCheckResize) throws IOException {
+ final int posHashCode = findBucket(hashCode);
+ // get the bucket for the given hash code
+ final int bucketArrayPos = posHashCode >> table.bucketsPerSegmentBits;
+ final int bucketInSegmentPos = (posHashCode & table.bucketsPerSegmentMask) << BUCKET_SIZE_BITS;
+ final MemorySegment bucket = this.buckets[bucketArrayPos];
+ return insertToBucket(bucket, bucketInSegmentPos, hashCode, pointer, spillingAllowed, sizeAddAndCheckResize);
+ }
+
+ /**
+ * Append record and insert to bucket.
+ */
+ boolean appendRecordAndInsert(BinaryRow record, int hashCode) throws IOException {
+ final int posHashCode = findBucket(hashCode);
+ // get the bucket for the given hash code
+ final int bucketArrayPos = posHashCode >> table.bucketsPerSegmentBits;
+ final int bucketInSegmentPos = (posHashCode & table.bucketsPerSegmentMask) << BUCKET_SIZE_BITS;
+ final MemorySegment bucket = this.buckets[bucketArrayPos];
+
+ if (!table.tryDistinctBuildRow ||
+ !partition.isInMemory() ||
+ !findFirstSameBuildRow(bucket, hashCode, bucketInSegmentPos, record)) {
+ int pointer = partition.insertIntoBuildBuffer(record);
+ if (pointer != -1) {
+ // record was inserted into an in-memory partition. a pointer must be inserted into the buckets
+ insertToBucket(bucket, bucketInSegmentPos, hashCode, pointer, true, true);
+ return true;
+ } else {
+ return false;
+ }
+ } else {
+ // distinct build rows in memory.
+ return true;
+ }
+ }
+
+ /**
+ * For distinct build.
+ */
+ private boolean findFirstSameBuildRow(
+ MemorySegment bucket,
+ int searchHashCode,
+ int bucketInSegmentOffset,
+ BinaryRow buildRowToInsert) {
+ int posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
+ int countInBucket = bucket.getShort(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
+ int numInBucket = 0;
+ RandomAccessInputView view = partition.getBuildStateInputView();
+ while (countInBucket != 0) {
+ while (numInBucket < countInBucket) {
+
+ final int thisCode = bucket.getInt(posInSegment);
+ posInSegment += HASH_CODE_LEN;
+
+ if (thisCode == searchHashCode) {
+ final int pointer = bucket.getInt(bucketInSegmentOffset +
+ BUCKET_POINTER_START_OFFSET + (numInBucket * POINTER_LEN));
+ numInBucket++;
+ try {
+ view.setReadPosition(pointer);
+ BinaryRow row = table.binaryBuildSideSerializer.mapFromPages(table.reuseBuildRow, view);
+ if (buildRowToInsert.equals(row)) {
+ return true;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Error deserializing key or value from the hashtable: " +
+ e.getMessage(), e);
+ }
+ } else {
+ numInBucket++;
+ }
+ }
+
+ // this segment is done. check if there is another chained bucket
+ final int forwardPointer = bucket.getInt(bucketInSegmentOffset + HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
+ return false;
+ }
+
+ final int overflowSegIndex = forwardPointer >>> table.segmentSizeBits;
+ bucket = overflowSegments[overflowSegIndex];
+ bucketInSegmentOffset = forwardPointer & table.segmentSizeMask;
+ countInBucket = bucket.getShort(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
+ posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
+ numInBucket = 0;
+ }
+ return false;
+ }
+
+ /**
+ * Probe start lookup joined build rows.
+ */
+ void startLookup(int hashCode) {
+ final int posHashCode = findBucket(hashCode);
+
+ // get the bucket for the given hash code
+ final int bucketArrayPos = posHashCode >> table.bucketsPerSegmentBits;
+ final int bucketInSegmentOffset = (posHashCode & table.bucketsPerSegmentMask) << BUCKET_SIZE_BITS;
+ final MemorySegment bucket = this.buckets[bucketArrayPos];
+ table.bucketIterator.set(bucket, overflowSegments, partition, hashCode, bucketInSegmentOffset);
+ }
+
+ void returnMemory(List<MemorySegment> target) {
+ target.addAll(Arrays.asList(overflowSegments).subList(0, numOverflowSegments));
+ target.addAll(Arrays.asList(buckets));
+ }
+
+ private void freeMemory() {
+ table.availableMemory.addAll(Arrays.asList(overflowSegments).subList(0, numOverflowSegments));
+ table.availableMemory.addAll(Arrays.asList(buckets));
+ }
+
+ /**
+ * Three situations:
+ * 1.Not use bloom filter, just free memory.
+ * 2.In rehash, free new memory and let rehash go build bloom filter from old memory.
+ * 3.Not in rehash and use bloom filter, build it and free memory.
+ */
+ void buildBloomFilterAndFree() {
+ if (inReHash || !table.useBloomFilters) {
+ freeMemory();
+ } else {
+ buildBloomFilterAndFree(buckets, numBuckets, overflowSegments);
+ }
+ }
+
+ private void buildBloomFilterAndFree(
+ MemorySegment[] buckets,
+ int numBuckets,
+ MemorySegment[] overflowSegments) {
+ if (table.useBloomFilters) {
+ long numRecords = (long) Math.max(partition.getBuildSideRecordCount() * 1.5, estimatedRowCount);
+
+ // BloomFilter size min of:
+ // 1.remain buffers
+ // 2.bf size for numRecords when fpp is 0.05
+ // 3.max init bucket area buffers.
+ int segSize = Math.min(
+ Math.min(table.remainBuffers(),
+ HashTableBloomFilter.optimalSegmentNumber(numRecords, table.pageSize(), 0.05)),
+ table.maxInitBufferOfBucketArea(table.partitionsBeingBuilt.size()));
+
+ if (segSize > 0) {
+ HashTableBloomFilter filter = new HashTableBloomFilter(
+ table.getNextBuffers(MathUtils.roundDownToPowerOf2(segSize)), numRecords);
+
+ // Add all records to bloom filter.
+ int scanCount = -1;
+ while (true) {
+ scanCount++;
+ if (scanCount >= numBuckets) {
+ break;
+ }
+ // move to next bucket, update all the current bucket status with new bucket information.
+ final int bucketArrayPos = scanCount >> table.bucketsPerSegmentBits;
+ int bucketInSegOffset = (scanCount & table.bucketsPerSegmentMask) << BUCKET_SIZE_BITS;
+ MemorySegment bucketSeg = buckets[bucketArrayPos];
+
+ int countInBucket = bucketSeg.getShort(bucketInSegOffset + HEADER_COUNT_OFFSET);
+ int numInBucket = 0;
+ while (countInBucket != 0) {
+ int hashCodeOffset = bucketInSegOffset + BUCKET_HEADER_LENGTH;
+ while (numInBucket < countInBucket) {
+ filter.addHash(bucketSeg.getInt(hashCodeOffset));
+ numInBucket++;
+ hashCodeOffset += HASH_CODE_LEN;
+ }
+
+ // this segment is done. check if there is another chained bucket
+ int forwardPointer = bucketSeg.getInt(bucketInSegOffset + HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
+ break;
+ }
+
+ final int overflowSegIndex = forwardPointer >>> table.segmentSizeBits;
+ bucketSeg = overflowSegments[overflowSegIndex];
+ bucketInSegOffset = forwardPointer & table.segmentSizeMask;
+ countInBucket = bucketSeg.getShort(bucketInSegOffset + HEADER_COUNT_OFFSET);
+ numInBucket = 0;
+ }
+ }
+
+ partition.bloomFilter = filter;
+ }
+ }
+
+ freeMemory(buckets, overflowSegments);
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashPartition.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashPartition.java
new file mode 100644
index 0000000..cc05301
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashPartition.java
@@ -0,0 +1,608 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentSource;
+import org.apache.flink.core.memory.SeekableDataInputView;
+import org.apache.flink.runtime.io.disk.RandomAccessInputView;
+import org.apache.flink.runtime.io.disk.iomanager.AbstractChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.BulkBlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.AbstractPagedInputView;
+import org.apache.flink.runtime.memory.AbstractPagedOutputView;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.runtime.compression.BlockCompressionFactory;
+import org.apache.flink.table.runtime.util.FileChannelUtil;
+import org.apache.flink.table.runtime.util.MemorySegmentPool;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.table.typeutils.BinaryRowSerializer;
+import org.apache.flink.util.MathUtils;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * A partition.
+ */
+public class BinaryHashPartition extends AbstractPagedInputView implements SeekableDataInputView {
+
+ private final BinaryRowSerializer buildSideSerializer;
+ private final BinaryRowSerializer probeSideSerializer;
+ private final int segmentSizeBits; // the number of bits in the mem segment size;
+
+ private boolean compressionEnable;
+ private BlockCompressionFactory compressionCodecFactory;
+ private int compressionBlockSize;
+
+ private final int memorySegmentSize; // the size of the memory segments being used
+ final int partitionNumber; // the number of the partition
+
+ long probeSideRecordCounter; // number of probe-side records in this partition
+ private MemorySegment[] partitionBuffers;
+ private int currentBufferNum;
+ private int finalBufferLimit;
+
+ private BuildSideBuffer buildSideWriteBuffer;
+ AbstractChannelWriterOutputView probeSideBuffer;
+ private long buildSideRecordCounter; // number of build-side records in this partition
+ private int recursionLevel; // the recursion level on which this partition lives
+
+ // the channel writer for the build side, if partition is spilled
+ private BlockChannelWriter<MemorySegment> buildSideChannel;
+
+ // bucket area of this partition
+ BinaryHashBucketArea bucketArea;
+
+ /**
+ * The bloom filter utility used to transform spilled partitions into a
+ * probabilistic filter.
+ */
+ HashTableBloomFilter bloomFilter;
+
+ private MemorySegmentPool memPool;
+
+ int probeNumBytesInLastSeg;
+
+ /**
+ * Creates a new partition, initially in memory, with one buffer for the build side. The
+ * partition is initialized to expect record insertions for the build side.
+ *
+ * @param partitionNumber The number of the partition.
+ * @param recursionLevel The recursion level - zero for partitions from the initial build,
+ * <i>n + 1</i> for partitions that are created from spilled partition
+ * with recursion level <i>n</i>.
+ * @param initialBuffer The initial buffer for this partition.
+ */
+ BinaryHashPartition(BinaryHashBucketArea bucketArea, BinaryRowSerializer buildSideAccessors, BinaryRowSerializer probeSideAccessors,
+ int partitionNumber, int recursionLevel, MemorySegment initialBuffer,
+ MemorySegmentPool memPool, int segmentSize, boolean compressionEnable,
+ BlockCompressionFactory compressionCodecFactory, int compressionBlockSize) {
+ super(0);
+ this.bucketArea = bucketArea;
+ this.buildSideSerializer = buildSideAccessors;
+ this.probeSideSerializer = probeSideAccessors;
+ this.partitionNumber = partitionNumber;
+ this.recursionLevel = recursionLevel;
+ this.memorySegmentSize = segmentSize;
+ this.segmentSizeBits = MathUtils.log2strict(segmentSize);
+ this.compressionEnable = compressionEnable;
+ this.compressionCodecFactory = compressionCodecFactory;
+ this.compressionBlockSize = compressionBlockSize;
+ this.buildSideWriteBuffer = new BuildSideBuffer(initialBuffer, memPool);
+ this.memPool = memPool;
+ }
+
+ /**
+ * Constructor creating a partition from a spilled partition file that could be read in one
+ * because it was known to completely fit into memory.
+ *
+ * @param buildSideAccessors The data type accessors for the build side data-type.
+ * @param probeSideAccessors The data type accessors for the probe side data-type.
+ * @param partitionNumber The number of the partition.
+ * @param recursionLevel The recursion level of the partition.
+ * @param buffers The memory segments holding the records.
+ * @param buildSideRecordCounter The number of records in the buffers.
+ * @param segmentSize The size of the memory segments.
+ */
+ BinaryHashPartition(BinaryHashBucketArea area, BinaryRowSerializer buildSideAccessors, BinaryRowSerializer probeSideAccessors,
+ int partitionNumber, int recursionLevel, List<MemorySegment> buffers,
+ long buildSideRecordCounter, int segmentSize, int lastSegmentLimit) {
+ super(0);
+ this.buildSideSerializer = buildSideAccessors;
+ this.probeSideSerializer = probeSideAccessors;
+ this.partitionNumber = partitionNumber;
+ this.recursionLevel = recursionLevel;
+
+ this.memorySegmentSize = segmentSize;
+ this.segmentSizeBits = MathUtils.log2strict(segmentSize);
+ this.finalBufferLimit = lastSegmentLimit;
+
+ this.partitionBuffers = buffers.toArray(new MemorySegment[buffers.size()]);
+ this.buildSideRecordCounter = buildSideRecordCounter;
+
+ this.bucketArea = area;
+ }
+
+ /**
+ * Gets the partition number of this partition.
+ *
+ * @return This partition's number.
+ */
+ int getPartitionNumber() {
+ return this.partitionNumber;
+ }
+
+ /**
+ * Gets this partition's recursion level.
+ *
+ * @return The partition's recursion level.
+ */
+ int getRecursionLevel() {
+ return this.recursionLevel;
+ }
+
+ /**
+ * Checks whether this partition is in memory or spilled.
+ *
+ * @return True, if the partition is in memory, false if it is spilled.
+ */
+ final boolean isInMemory() {
+ return this.buildSideChannel == null;
+ }
+
+ /**
+ * Gets the number of memory segments used by this partition, which includes build side
+ * memory buffers and overflow memory segments.
+ *
+ * @return The number of occupied memory segments.
+ */
+ int getNumOccupiedMemorySegments() {
+ // either the number of memory segments, or one for spilling
+ final int numPartitionBuffers = this.partitionBuffers != null ?
+ this.partitionBuffers.length
+ : this.buildSideWriteBuffer.getNumOccupiedMemorySegments();
+ return numPartitionBuffers + bucketArea.buckets.length + bucketArea.numOverflowSegments;
+ }
+
+ int getBuildSideBlockCount() {
+ return this.partitionBuffers == null ? this.buildSideWriteBuffer.getBlockCount()
+ : this.partitionBuffers.length;
+ }
+
+ RandomAccessInputView getBuildStateInputView() {
+ return this.buildSideWriteBuffer.getBuildStageInputView();
+ }
+
+ int getProbeSideBlockCount() {
+ return this.probeSideBuffer == null ? -1 : this.probeSideBuffer.getBlockCount();
+ }
+
+ long getBuildSideRecordCount() {
+ return this.buildSideRecordCounter;
+ }
+
+ BlockChannelWriter<MemorySegment> getBuildSideChannel() {
+ return this.buildSideChannel;
+ }
+
+ boolean testHashBloomFilter(int hash) {
+ return bloomFilter == null || bloomFilter.testHash(hash);
+ }
+
+ private void freeBloomFilter() {
+ memPool.returnAll(Arrays.asList(bloomFilter.getBuffers()));
+ bloomFilter = null;
+ }
+
+ /**
+ * Add new hash to bloomFilter when insert a record to spilled partition.
+ */
+ void addHashBloomFilter(int hash) {
+ if (bloomFilter != null) {
+ // check if too full.
+ if (!bloomFilter.addHash(hash)) {
+ freeBloomFilter();
+ }
+ }
+ }
+
+ /**
+ * Inserts the given object into the current buffer. This method returns a pointer that
+ * can be used to address the written record in this partition, if it is in-memory.
+ * The returned pointers have no expressiveness in the case where the partition is spilled.
+ *
+ * @param record The object to be written to the partition.
+ * @return A pointer to the object in the partition, or <code>-1</code>, if the partition is
+ * spilled.
+ * @throws IOException Thrown, when this is a spilled partition and the write failed.
+ */
+ final int insertIntoBuildBuffer(BinaryRow record) throws IOException {
+ this.buildSideRecordCounter++;
+
+ if (isInMemory()) {
+ final long pointer = this.buildSideWriteBuffer.getCurrentPointer();
+ int skip = this.buildSideSerializer.serializeToPages(record, this.buildSideWriteBuffer);
+ if (isInMemory()) {
+ long ret = pointer + skip;
+ if (ret > Integer.MAX_VALUE) {
+ throw new RuntimeException("Too more data in this partition: " + ret);
+ }
+ return (int) ret;
+ } else {
+ return -1;
+ }
+ } else {
+ this.buildSideSerializer.serializeToPages(record, this.buildSideWriteBuffer);
+ return -1;
+ }
+ }
+
+ /**
+ * Inserts the given record into the probe side buffers. This method is only applicable when
+ * the partition was spilled while processing the build side.
+ *
+ * <p>If this method is invoked when the partition is still being built, it has undefined
+ * behavior.
+ *
+ * @param record The record to be inserted into the probe side buffers.
+ * @throws IOException Thrown, if the buffer is full, needs to be spilled, and spilling causes
+ * an error.
+ */
+ final void insertIntoProbeBuffer(BinaryRow record) throws IOException {
+ this.probeSideSerializer.serialize(record, this.probeSideBuffer);
+ this.probeSideRecordCounter++;
+ }
+
+ /**
+ * Spills this partition to disk and sets it up such that it continues spilling records that are
+ * added to
+ * it. The spilling process must free at least one buffer, either in the partition's record
+ * buffers, or in
+ * the memory segments for overflow buckets.
+ * The partition immediately takes back one buffer to use it for further spilling.
+ *
+ * @param ioAccess The I/O manager to be used to create a writer to disk.
+ * @param targetChannel The id of the target channel for this partition.
+ * @return The number of buffers that were freed by spilling this partition.
+ * @throws IOException Thrown, if the writing failed.
+ */
+ int spillPartition(
+ IOManager ioAccess, FileIOChannel.ID targetChannel,
+ LinkedBlockingQueue<MemorySegment> bufferReturnQueue) throws IOException {
+ // sanity checks
+ if (!isInMemory()) {
+ throw new RuntimeException("Bug in Hybrid Hash Join: " +
+ "Request to spill a partition that has already been spilled.");
+ }
+ if (getNumOccupiedMemorySegments() < 2) {
+ throw new RuntimeException("Bug in Hybrid Hash Join: " +
+ "Request to spill a partition with less than two buffers.");
+ }
+
+ // create the channel block writer and spill the current buffers
+ // that keep the build side buffers current block, as it is most likely not full, yet
+ // we return the number of blocks that become available
+ this.buildSideChannel = FileChannelUtil.createBlockChannelWriter(
+ ioAccess,
+ targetChannel,
+ bufferReturnQueue,
+ compressionEnable,
+ compressionCodecFactory,
+ compressionBlockSize,
+ memorySegmentSize);
+ return this.buildSideWriteBuffer.spill(this.buildSideChannel);
+ }
+
+ void buildBloomFilterAndFreeBucket() {
+ if (bucketArea != null) {
+ this.bucketArea.buildBloomFilterAndFree();
+ this.bucketArea = null;
+ }
+ }
+
+ /**
+ * After build phase.
+ * @return build spill return buffer, if have spilled, it returns the current write buffer,
+ * because it was used all the time in build phase, so it can only be returned at this time.
+ */
+ int finalizeBuildPhase(IOManager ioAccess, FileIOChannel.Enumerator probeChannelEnumerator) throws IOException {
+ this.finalBufferLimit = this.buildSideWriteBuffer.getCurrentPositionInSegment();
+ this.partitionBuffers = this.buildSideWriteBuffer.close();
+
+ if (!isInMemory()) {
+ // close the channel.
+ this.buildSideChannel.close();
+
+ this.probeSideBuffer = FileChannelUtil.createOutputView(ioAccess, probeChannelEnumerator.next(),
+ compressionEnable, compressionCodecFactory, compressionBlockSize, memorySegmentSize);
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ /**
+ * @param keepUnprobedSpilledPartitions If true then partitions that were spilled but received
+ * no further probe
+ * requests will be retained; used for build-side outer
+ * joins.
+ */
+ void finalizeProbePhase(List<MemorySegment> freeMemory, List<BinaryHashPartition> spilledPartitions,
+ boolean keepUnprobedSpilledPartitions) throws IOException {
+ if (isInMemory()) {
+ this.bucketArea.returnMemory(freeMemory);
+ this.bucketArea = null;
+ // return the partition buffers
+ Collections.addAll(freeMemory, this.partitionBuffers);
+ this.partitionBuffers = null;
+ } else {
+ if (bloomFilter != null) {
+ freeBloomFilter();
+ }
+ if (this.probeSideRecordCounter == 0 && !keepUnprobedSpilledPartitions) {
+ // delete the spill files
+ this.probeSideBuffer.close();
+ this.buildSideChannel.deleteChannel();
+ this.probeSideBuffer.getChannel().deleteChannel();
+ } else {
+ // flush the last probe side buffer and register this partition as pending
+ probeNumBytesInLastSeg = this.probeSideBuffer.close();
+ spilledPartitions.add(this);
+ }
+ }
+ }
+
+ void clearAllMemory(List<MemorySegment> target) {
+ // return current buffers from build side and probe side
+ if (this.buildSideWriteBuffer != null) {
+ if (this.buildSideWriteBuffer.getCurrentSegment() != null) {
+ target.add(this.buildSideWriteBuffer.getCurrentSegment());
+ }
+ target.addAll(this.buildSideWriteBuffer.targetList);
+ this.buildSideWriteBuffer.targetList.clear();
+ this.buildSideWriteBuffer = null;
+ }
+
+ // return the overflow segments
+ if (this.bucketArea != null) {
+ bucketArea.returnMemory(target);
+ }
+
+ if (bloomFilter != null) {
+ freeBloomFilter();
+ }
+
+ // return the partition buffers
+ if (this.partitionBuffers != null) {
+ Collections.addAll(target, this.partitionBuffers);
+ this.partitionBuffers = null;
+ }
+
+ // clear the channels
+ try {
+ if (this.buildSideChannel != null) {
+ this.buildSideChannel.close();
+ this.buildSideChannel.deleteChannel();
+ }
+ if (this.probeSideBuffer != null) {
+ this.probeSideBuffer.close();
+ this.probeSideBuffer.getChannel().deleteChannel();
+ this.probeSideBuffer = null;
+ }
+ } catch (IOException ioex) {
+ throw new RuntimeException("Error deleting the partition files. " +
+ "Some temporary files might not be removed.", ioex);
+ }
+ }
+
+ final PartitionIterator newPartitionIterator() {
+ return new PartitionIterator();
+ }
+
+ final int getLastSegmentLimit() {
+ return this.finalBufferLimit;
+ }
+
+ @Override
+ public void setReadPosition(long pointer) {
+ final int bufferNum = (int) (pointer >>> this.segmentSizeBits);
+ final int offset = (int) (pointer & (this.memorySegmentSize - 1));
+
+ this.currentBufferNum = bufferNum;
+ seekInput(this.partitionBuffers[bufferNum], offset,
+ bufferNum < this.partitionBuffers.length - 1 ? this.memorySegmentSize
+ : this.finalBufferLimit);
+
+ }
+
+ @Override
+ protected MemorySegment nextSegment(MemorySegment current) throws IOException {
+ this.currentBufferNum++;
+ if (this.currentBufferNum < this.partitionBuffers.length) {
+ return this.partitionBuffers[this.currentBufferNum];
+ } else {
+ throw new EOFException();
+ }
+ }
+
+ @Override
+ protected int getLimitForSegment(MemorySegment segment) {
+ return segment == this.partitionBuffers[partitionBuffers.length - 1]
+ ? this.finalBufferLimit : this.memorySegmentSize;
+ }
+
+ /**
+ * Build side buffer.
+ */
+ protected static final class BuildSideBuffer extends AbstractPagedOutputView {
+
+ private final ArrayList<MemorySegment> targetList;
+ /**
+ * Segments of in memory partition, include the current segment.
+ * Cleared after spilled.
+ */
+ private final ArrayList<MemorySegment> buildStageSegments;
+ private final RandomAccessInputView buildStageInputView;
+
+ private final MemorySegmentSource memSource;
+ private final int sizeBits;
+ private BlockChannelWriter<MemorySegment> writer;
+ private int currentBlockNumber;
+
+ private BuildSideBuffer(MemorySegment initialSegment, MemorySegmentSource memSource) {
+ super(initialSegment, initialSegment.size(), 0);
+
+ this.memSource = memSource;
+ this.sizeBits = MathUtils.log2strict(initialSegment.size());
+ this.targetList = new ArrayList<>();
+ this.buildStageSegments = new ArrayList<>();
+ this.buildStageSegments.add(initialSegment);
+ this.buildStageInputView = new RandomAccessInputView(
+ buildStageSegments, initialSegment.size());
+ }
+
+ @Override
+ protected MemorySegment nextSegment(
+ MemorySegment current,
+ int bytesUsed) throws IOException {
+ final MemorySegment next;
+ if (this.writer == null) {
+ // Must first add current segment:
+ // This may happen when you need to spill:
+ // A partition called nextSegment, can not get memory, need to spill, the result
+ // give itself to the spill, Since it is switching currentSeg, it is necessary
+ // to give the previous currSeg to spill.
+ this.targetList.add(current);
+ next = this.memSource.nextSegment();
+ buildStageSegments.add(next);
+ } else {
+ this.writer.writeBlock(current);
+ try {
+ next = this.writer.getReturnQueue().take();
+ } catch (InterruptedException iex) {
+ throw new IOException("Hash Join Partition was interrupted while " +
+ "grabbing a new write-behind buffer.");
+ }
+ }
+
+ this.currentBlockNumber++;
+ return next;
+ }
+
+ RandomAccessInputView getBuildStageInputView() {
+ return buildStageInputView;
+ }
+
+ long getCurrentPointer() {
+ return (((long) this.currentBlockNumber) << this.sizeBits)
+ + getCurrentPositionInSegment();
+ }
+
+ int getBlockCount() {
+ return this.currentBlockNumber + 1;
+ }
+
+ int getNumOccupiedMemorySegments() {
+ // return the current segment + all filled segments
+ return this.targetList.size() + 1;
+ }
+
+ int spill(BlockChannelWriter<MemorySegment> writer) throws IOException {
+ this.writer = writer;
+ final int numSegments = this.targetList.size();
+ for (MemorySegment segment : this.targetList) {
+ this.writer.writeBlock(segment);
+ }
+ this.targetList.clear();
+ return numSegments;
+ }
+
+ MemorySegment[] close() throws IOException {
+ final MemorySegment current = getCurrentSegment();
+ if (current == null) {
+ throw new IllegalStateException("Illegal State in HashPartition: " +
+ "No current buffer when finalizing build side.");
+ }
+ clear();
+
+ if (this.writer == null) {
+ this.targetList.add(current);
+ MemorySegment[] buffers =
+ this.targetList.toArray(new MemorySegment[this.targetList.size()]);
+ this.targetList.clear();
+ this.buildStageSegments.clear();
+ return buffers;
+ } else {
+ writer.writeBlock(current);
+ return null;
+ }
+ }
+ }
+
+ /**
+ * For spilled partition to rebuild index and hashcode when memory can
+ * store all the build side data.
+ * (After bulk load to memory, see {@link BulkBlockChannelReader}).
+ */
+ final class PartitionIterator implements RowIterator<BinaryRow> {
+
+ private long currentPointer;
+
+ private BinaryRow reuse;
+
+ private PartitionIterator() {
+ this.reuse = buildSideSerializer.createInstance();
+ setReadPosition(0);
+ }
+
+ @Override
+ public boolean advanceNext() {
+ final int pos = getCurrentPositionInSegment();
+ final int buffer = BinaryHashPartition.this.currentBufferNum;
+
+ this.currentPointer = (((long) buffer) << segmentSizeBits) + pos;
+
+ try {
+ reuse = buildSideSerializer.mapFromPages(reuse, BinaryHashPartition.this);
+ return true;
+ } catch (EOFException e) {
+ return false;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ final long getPointer() {
+ return this.currentPointer;
+ }
+
+ @Override
+ public BinaryRow getRow() {
+ return this.reuse;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashTable.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashTable.java
new file mode 100644
index 0000000..dc8da66
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryHashTable.java
@@ -0,0 +1,666 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.util.BitSet;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.util.BinaryRowUtil;
+import org.apache.flink.table.generated.JoinCondition;
+import org.apache.flink.table.generated.Projection;
+import org.apache.flink.table.runtime.io.ChannelWithMeta;
+import org.apache.flink.table.runtime.join.HashJoinType;
+import org.apache.flink.table.runtime.join.NullAwareJoinHelper;
+import org.apache.flink.table.runtime.util.FileChannelUtil;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.table.typeutils.AbstractRowSerializer;
+import org.apache.flink.table.typeutils.BinaryRowSerializer;
+import org.apache.flink.util.MathUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * An implementation of a Hybrid Hash Join. The join starts operating in memory and gradually
+ * starts spilling contents to disk, when the memory is not sufficient. It does not need to know a
+ * priority how large the input will be.
+ *
+ * <p>The design of this class follows in many parts the design presented in
+ * "Hash joins and hash teams in Microsoft SQL Server", by Goetz Graefe et al. In its current state,
+ * the implementation lacks features like dynamic role reversal, partition tuning, or histogram
+ * guided partitioning.</p>
+ */
+public class BinaryHashTable extends BaseHybridHashTable {
+
+ /**
+ * The utilities to serialize the build side data types.
+ */
+ final BinaryRowSerializer binaryBuildSideSerializer;
+
+ private final AbstractRowSerializer originBuildSideSerializer;
+
+ /**
+ * The utilities to serialize the probe side data types.
+ */
+ private final BinaryRowSerializer binaryProbeSideSerializer;
+
+ private final AbstractRowSerializer originProbeSideSerializer;
+
+ /**
+ * The utilities to hash and compare the build side data types.
+ */
+ private final Projection<BaseRow, BinaryRow> buildSideProjection;
+
+ /**
+ * The utilities to hash and compare the probe side data types.
+ */
+ private final Projection<BaseRow, BinaryRow> probeSideProjection;
+
+ final int bucketsPerSegment;
+
+ /**
+ * The number of hash table buckets in a single memory segment - 1.
+ * Because memory segments can be comparatively large, we fit multiple buckets into one memory
+ * segment.
+ * This variable is a mask that is 1 in the lower bits that define the number of a bucket
+ * in a segment.
+ */
+ final int bucketsPerSegmentMask;
+
+ /**
+ * The number of bits that describe the position of a bucket in a memory segment. Computed as
+ * log2(bucketsPerSegment).
+ */
+ final int bucketsPerSegmentBits;
+
+ /** Flag to enable/disable bloom filters for spilled partitions. */
+ final boolean useBloomFilters;
+
+ /**
+ * The partitions that are built by processing the current partition.
+ */
+ final ArrayList<BinaryHashPartition> partitionsBeingBuilt;
+
+ /**
+ * BitSet which used to mark whether the element(int build side) has successfully matched during
+ * probe phase. As there are 9 elements in each bucket, we assign 2 bytes to BitSet.
+ */
+ final BitSet probedSet = new BitSet(2);
+
+ /**
+ * The partitions that have been spilled previously and are pending to be processed.
+ */
+ private final ArrayList<BinaryHashPartition> partitionsPending;
+
+ private final JoinCondition condFunc;
+
+ private final boolean reverseJoin;
+
+ /**
+ * Should filter null keys.
+ */
+ private final int[] nullFilterKeys;
+
+ /**
+ * No keys need to filter null.
+ */
+ private final boolean nullSafe;
+
+ /**
+ * Filter null to all keys.
+ */
+ private final boolean filterAllNulls;
+
+ /**
+ * Iterator over the elements in the hash table.
+ */
+ LookupBucketIterator bucketIterator;
+
+ /**
+ * Iterator over the elements from the probe side.
+ */
+ private ProbeIterator probeIterator;
+
+ final HashJoinType type;
+
+ private RowIterator<BinaryRow> buildIterator;
+
+ private boolean probeMatchedPhase = true;
+
+ private boolean buildIterVisited = false;
+
+ private BinaryRow probeKey;
+ private BaseRow probeRow;
+
+ BinaryRow reuseBuildRow;
+
+ public BinaryHashTable(
+ Configuration conf,
+ Object owner,
+ AbstractRowSerializer buildSideSerializer,
+ AbstractRowSerializer probeSideSerializer,
+ Projection<BaseRow, BinaryRow> buildSideProjection,
+ Projection<BaseRow, BinaryRow> probeSideProjection,
+ MemoryManager memManager,
+ long reservedMemorySize,
+ IOManager ioManager,
+ int avgRecordLen,
+ int buildRowCount,
+ boolean useBloomFilters,
+ HashJoinType type,
+ JoinCondition condFunc,
+ boolean reverseJoin,
+ boolean[] filterNulls,
+ boolean tryDistinctBuildRow) {
+ this(conf, owner, buildSideSerializer, probeSideSerializer, buildSideProjection, probeSideProjection, memManager,
+ reservedMemorySize, reservedMemorySize, 0, ioManager, avgRecordLen, buildRowCount, useBloomFilters, type, condFunc,
+ reverseJoin, filterNulls, tryDistinctBuildRow);
+ }
+
+ public BinaryHashTable(
+ Configuration conf,
+ Object owner,
+ AbstractRowSerializer buildSideSerializer,
+ AbstractRowSerializer probeSideSerializer,
+ Projection<BaseRow, BinaryRow> buildSideProjection,
+ Projection<BaseRow, BinaryRow> probeSideProjection,
+ MemoryManager memManager,
+ long reservedMemorySize,
+ long preferredMemorySize,
+ long perRequestMemorySize,
+ IOManager ioManager,
+ int avgRecordLen,
+ long buildRowCount,
+ boolean useBloomFilters,
+ HashJoinType type,
+ JoinCondition condFunc,
+ boolean reverseJoin,
+ boolean[] filterNulls,
+ boolean tryDistinctBuildRow) {
+ super(conf, owner, memManager, reservedMemorySize, preferredMemorySize, perRequestMemorySize,
+ ioManager, avgRecordLen, buildRowCount, !type.buildLeftSemiOrAnti() && tryDistinctBuildRow);
+ // assign the members
+ this.originBuildSideSerializer = buildSideSerializer;
+ this.binaryBuildSideSerializer = new BinaryRowSerializer(buildSideSerializer.getArity());
+ this.reuseBuildRow = binaryBuildSideSerializer.createInstance();
+ this.originProbeSideSerializer = probeSideSerializer;
+ this.binaryProbeSideSerializer = new BinaryRowSerializer(originProbeSideSerializer.getArity());
+
+ this.buildSideProjection = buildSideProjection;
+ this.probeSideProjection = probeSideProjection;
+ this.useBloomFilters = useBloomFilters;
+ this.type = type;
+ this.condFunc = condFunc;
+ this.reverseJoin = reverseJoin;
+ this.nullFilterKeys = NullAwareJoinHelper.getNullFilterKeys(filterNulls);
+ this.nullSafe = nullFilterKeys.length == 0;
+ this.filterAllNulls = nullFilterKeys.length == filterNulls.length;
+
+ this.bucketsPerSegment = this.segmentSize >> BinaryHashBucketArea.BUCKET_SIZE_BITS;
+ checkArgument(bucketsPerSegment != 0,
+ "Hash Table requires buffers of at least " + BinaryHashBucketArea.BUCKET_SIZE + " bytes.");
+ this.bucketsPerSegmentMask = bucketsPerSegment - 1;
+ this.bucketsPerSegmentBits = MathUtils.log2strict(bucketsPerSegment);
+
+ this.partitionsBeingBuilt = new ArrayList<>();
+ this.partitionsPending = new ArrayList<>();
+
+ createPartitions(initPartitionFanOut, 0);
+ }
+
+ // ========================== build phase public method ======================================
+
+ /**
+ * Put a build side row to hash table.
+ */
+ public void putBuildRow(BaseRow row) throws IOException {
+ final int hashCode = hash(this.buildSideProjection.apply(row).hashCode(), 0);
+ // TODO: combine key projection and build side conversion to code gen.
+ insertIntoTable(originBuildSideSerializer.baseRowToBinary(row), hashCode);
+ }
+
+ /**
+ * End build phase.
+ */
+ public void endBuild() throws IOException {
+ // finalize the partitions
+ int buildWriteBuffers = 0;
+ for (BinaryHashPartition p : this.partitionsBeingBuilt) {
+ buildWriteBuffers += p.finalizeBuildPhase(this.ioManager, this.currentEnumerator);
+ }
+ buildSpillRetBufferNumbers += buildWriteBuffers;
+
+ // the first prober is the probe-side input, but the input is null at beginning
+ this.probeIterator = new ProbeIterator(this.binaryProbeSideSerializer.createInstance());
+
+ // the bucket iterator can remain constant over the time
+ this.bucketIterator = new LookupBucketIterator(this);
+ }
+
+ // ========================== probe phase public method ======================================
+
+ /**
+ * Find matched build side rows for a probe row.
+ * @return return false if the target partition has spilled, we will spill this probe row too.
+ * The row will be re-match in rebuild phase.
+ */
+ public boolean tryProbe(BaseRow record) throws IOException {
+ if (!this.probeIterator.hasSource()) {
+ // set the current probe value when probeIterator is null at the begging.
+ this.probeIterator.setInstance(record);
+ }
+ // calculate the hash
+ BinaryRow probeKey = probeSideProjection.apply(record);
+ final int hash = hash(probeKey.hashCode(), this.currentRecursionDepth);
+
+ BinaryHashPartition p = this.partitionsBeingBuilt.get(hash % partitionsBeingBuilt.size());
+
+ // for an in-memory partition, process set the return iterators, else spill the probe records
+ if (p.isInMemory()) {
+ this.probeKey = probeKey;
+ this.probeRow = record;
+ p.bucketArea.startLookup(hash);
+ return true;
+ } else {
+ if (p.testHashBloomFilter(hash)) {
+ BinaryRow row = originProbeSideSerializer.baseRowToBinary(record);
+ p.insertIntoProbeBuffer(row);
+ }
+ return false;
+ }
+ }
+
+ // ========================== rebuild phase public method ======================================
+
+ /**
+ * Next record from rebuilt spilled partition or build side outer partition.
+ */
+ public boolean nextMatching() throws IOException {
+ if (type.needSetProbed()) {
+ return processProbeIter() || processBuildIter() || prepareNextPartition();
+ } else {
+ return processProbeIter() || prepareNextPartition();
+ }
+ }
+
+ public BaseRow getCurrentProbeRow() {
+ if (this.probeMatchedPhase) {
+ return this.probeIterator.current();
+ } else {
+ return null;
+ }
+ }
+
+ public RowIterator<BinaryRow> getBuildSideIterator() {
+ return probeMatchedPhase ? bucketIterator : buildIterator;
+ }
+
+ // ================================ internal method ===========================================
+
+ /**
+ * Determines the number of buffers to be used for asynchronous write behind. It is currently
+ * computed as the logarithm of the number of buffers to the base 4, rounded up, minus 2.
+ * The upper limit for the number of write behind buffers is however set to six.
+ *
+ * @param numBuffers The number of available buffers.
+ * @return The number
+ */
+ @VisibleForTesting
+ static int getNumWriteBehindBuffers(int numBuffers) {
+ int numIOBufs = (int) (Math.log(numBuffers) / Math.log(4) - 1.5);
+ return numIOBufs > 6 ? 6 : numIOBufs;
+ }
+
+ private boolean processProbeIter() throws IOException {
+
+ // the prober's source is null at the begging.
+ if (this.probeIterator.hasSource()) {
+ final ProbeIterator probeIter = this.probeIterator;
+
+ if (!this.probeMatchedPhase) {
+ return false;
+ }
+
+ BinaryRow next;
+ while ((next = probeIter.next()) != null) {
+ BinaryRow probeKey = probeSideProjection.apply(next);
+ final int hash = hash(probeKey.hashCode(), this.currentRecursionDepth);
+
+ final BinaryHashPartition p = this.partitionsBeingBuilt.get(hash % partitionsBeingBuilt.size());
+
+ // for an in-memory partition, process set the return iterators, else spill the probe records
+ if (p.isInMemory()) {
+ this.probeKey = probeKey;
+ this.probeRow = next;
+ p.bucketArea.startLookup(hash);
+ return true;
+ } else {
+ p.insertIntoProbeBuffer(next);
+ }
+ }
+ // -------------- partition done ---------------
+
+ return false;
+ } else {
+ return false;
+ }
+ }
+
+ private boolean processBuildIter() throws IOException {
+ if (this.buildIterVisited) {
+ return false;
+ }
+
+ this.probeMatchedPhase = false;
+ this.buildIterator = new BuildSideIterator(
+ this.binaryBuildSideSerializer, reuseBuildRow,
+ this.partitionsBeingBuilt, probedSet, type.equals(HashJoinType.BUILD_LEFT_SEMI));
+ this.buildIterVisited = true;
+ return true;
+ }
+
+ private boolean prepareNextPartition() throws IOException {
+ // finalize and cleanup the partitions of the current table
+ for (final BinaryHashPartition p : this.partitionsBeingBuilt) {
+ p.finalizeProbePhase(this.availableMemory, this.partitionsPending, type.needSetProbed());
+ }
+
+ this.partitionsBeingBuilt.clear();
+
+ if (this.currentSpilledBuildSide != null) {
+ this.currentSpilledBuildSide.getChannel().closeAndDelete();
+ this.currentSpilledBuildSide = null;
+ }
+
+ if (this.currentSpilledProbeSide != null) {
+ this.currentSpilledProbeSide.getChannel().closeAndDelete();
+ this.currentSpilledProbeSide = null;
+ }
+
+ if (this.partitionsPending.isEmpty()) {
+ // no more data
+ return false;
+ }
+
+ // there are pending partitions
+ final BinaryHashPartition p = this.partitionsPending.get(0);
+ LOG.info(String.format("Begin to process spilled partition [%d]", p.getPartitionNumber()));
+
+ if (p.probeSideRecordCounter == 0) {
+ // unprobed spilled partitions are only re-processed for a build-side outer join;
+ // there is no need to create a hash table since there are no probe-side records
+ this.currentSpilledBuildSide = createInputView(p.getBuildSideChannel().getChannelID(),
+ p.getBuildSideBlockCount(), p.getLastSegmentLimit());
+ this.buildIterator = new WrappedRowIterator<>(
+ new BinaryRowChannelInputViewIterator(currentSpilledBuildSide, this.binaryBuildSideSerializer),
+ binaryBuildSideSerializer.createInstance());
+
+ this.partitionsPending.remove(0);
+
+ return true;
+ }
+
+ this.probeMatchedPhase = true;
+ this.buildIterVisited = false;
+
+ // build the next table; memory must be allocated after this call
+ buildTableFromSpilledPartition(p);
+
+ // set the probe side
+ ChannelWithMeta channelWithMeta = new ChannelWithMeta(
+ p.probeSideBuffer.getChannel().getChannelID(),
+ p.probeSideBuffer.getBlockCount(),
+ p.probeNumBytesInLastSeg);
+ this.currentSpilledProbeSide = FileChannelUtil.createInputView(
+ ioManager, channelWithMeta, new ArrayList<>(),
+ compressionEnable, compressionCodecFactory, compressionBlockSize, segmentSize);
+
+ ChannelReaderInputViewIterator<BinaryRow> probeReader =
+ new ChannelReaderInputViewIterator(
+ this.currentSpilledProbeSide, new ArrayList<>(), this.binaryProbeSideSerializer);
+ this.probeIterator.set(probeReader);
+ this.probeIterator.setReuse(binaryProbeSideSerializer.createInstance());
+
+ // unregister the pending partition
+ this.partitionsPending.remove(0);
+ this.currentRecursionDepth = p.getRecursionLevel() + 1;
+
+ // recursively get the next
+ return nextMatching();
+ }
+
+ private void buildTableFromSpilledPartition(
+ final BinaryHashPartition p) throws IOException {
+
+ final int nextRecursionLevel = p.getRecursionLevel() + 1;
+ if (nextRecursionLevel == 2) {
+ LOG.info("Recursive hash join: partition number is " + p.getPartitionNumber());
+ } else if (nextRecursionLevel > MAX_RECURSION_DEPTH) {
+ throw new RuntimeException("Hash join exceeded maximum number of recursions, without reducing "
+ + "partitions enough to be memory resident. Probably cause: Too many duplicate keys.");
+ }
+
+ if (p.getBuildSideBlockCount() > p.getProbeSideBlockCount()) {
+ LOG.info(String.format(
+ "Hash join: Partition(%d) " +
+ "build side block [%d] more than probe side block [%d]",
+ p.getPartitionNumber(),
+ p.getBuildSideBlockCount(),
+ p.getProbeSideBlockCount()));
+ }
+
+ // we distinguish two cases here:
+ // 1) The partition fits entirely into main memory. That is the case if we have enough buffers for
+ // all partition segments, plus enough buffers to hold the table structure.
+ // --> We read the partition in as it is and create a hashtable that references only
+ // that single partition.
+ // 2) We can not guarantee that enough memory segments are available and read the partition
+ // in, distributing its data among newly created partitions.
+ final int totalBuffersAvailable = this.availableMemory.size() + this.buildSpillRetBufferNumbers;
+ if (totalBuffersAvailable != this.reservedNumBuffers + this.allocatedFloatingNum) {
+ throw new RuntimeException(String.format("Hash Join bug in memory management: Memory buffers leaked." +
+ " availableMemory(%s), buildSpillRetBufferNumbers(%s), reservedNumBuffers(%s), allocatedFloatingNum(%s)",
+ availableMemory.size(), buildSpillRetBufferNumbers, reservedNumBuffers, allocatedFloatingNum));
+ }
+
+ long numBuckets = p.getBuildSideRecordCount() / BinaryHashBucketArea.NUM_ENTRIES_PER_BUCKET + 1;
+
+ // we need to consider the worst case where everything hashes to one bucket which needs to overflow by the same
+ // number of total buckets again. Also, one buffer needs to remain for the probing
+ int maxBucketAreaBuffers = Math.max((int) (2 * (numBuckets / (this.bucketsPerSegmentMask + 1))), 1);
+ final long totalBuffersNeeded = maxBucketAreaBuffers + p.getBuildSideBlockCount() + 2;
+
+ if (totalBuffersNeeded < totalBuffersAvailable) {
+ LOG.info(String.format("Build in memory hash table from spilled partition [%d]", p.getPartitionNumber()));
+
+ // first read the partition in
+ final List<MemorySegment> partitionBuffers = readAllBuffers(p.getBuildSideChannel().getChannelID(), p.getBuildSideBlockCount());
+ BinaryHashBucketArea area = new BinaryHashBucketArea(this, (int) p.getBuildSideRecordCount(), maxBucketAreaBuffers);
+ final BinaryHashPartition newPart = new BinaryHashPartition(area, this.binaryBuildSideSerializer, this.binaryProbeSideSerializer,
+ 0, nextRecursionLevel, partitionBuffers, p.getBuildSideRecordCount(), this.segmentSize, p.getLastSegmentLimit());
+ area.setPartition(newPart);
+
+ this.partitionsBeingBuilt.add(newPart);
+
+ // now, index the partition through a hash table
+ final BinaryHashPartition.PartitionIterator pIter = newPart.newPartitionIterator();
+
+ while (pIter.advanceNext()) {
+ final int hashCode = hash(buildSideProjection.apply(pIter.getRow()).hashCode(), nextRecursionLevel);
+ final int pointer = (int) pIter.getPointer();
+ area.insertToBucket(hashCode, pointer, false, true);
+ }
+ } else {
+ // go over the complete input and insert every element into the hash table
+ // compute in how many splits, we'd need to partition the result
+ final int splits = (int) (totalBuffersNeeded / totalBuffersAvailable) + 1;
+ final int partitionFanOut = Math.min(Math.min(10 * splits, MAX_NUM_PARTITIONS), maxNumPartition());
+
+ createPartitions(partitionFanOut, nextRecursionLevel);
+ LOG.info(String.format("Build hybrid hash table from spilled partition [%d] with recursion level [%d]",
+ p.getPartitionNumber(), nextRecursionLevel));
+
+ ChannelReaderInputView inView = createInputView(p.getBuildSideChannel().getChannelID(), p.getBuildSideBlockCount(), p.getLastSegmentLimit());
+ final BinaryRowChannelInputViewIterator inIter =
+ new BinaryRowChannelInputViewIterator(inView, this.binaryBuildSideSerializer);
+ BinaryRow rec = this.binaryBuildSideSerializer.createInstance();
+ while ((rec = inIter.next(rec)) != null) {
+ final int hashCode = hash(this.buildSideProjection.apply(rec).hashCode(), nextRecursionLevel);
+ insertIntoTable(rec, hashCode);
+ }
+
+ inView.getChannel().closeAndDelete();
+
+ // finalize the partitions
+ int buildWriteBuffers = 0;
+ for (BinaryHashPartition part : this.partitionsBeingBuilt) {
+ buildWriteBuffers += part.finalizeBuildPhase(this.ioManager, this.currentEnumerator);
+ }
+ buildSpillRetBufferNumbers += buildWriteBuffers;
+ }
+ }
+
+ private void insertIntoTable(final BinaryRow record, final int hashCode) throws IOException {
+ BinaryHashPartition p = partitionsBeingBuilt.get(hashCode % partitionsBeingBuilt.size());
+ if (p.isInMemory()) {
+ if (!p.bucketArea.appendRecordAndInsert(record, hashCode)) {
+ p.addHashBloomFilter(hashCode);
+ }
+ } else {
+ p.insertIntoBuildBuffer(record);
+ p.addHashBloomFilter(hashCode);
+ }
+ }
+
+ private void createPartitions(int numPartitions, int recursionLevel) {
+ // sanity check
+ ensureNumBuffersReturned(numPartitions);
+
+ this.currentEnumerator = this.ioManager.createChannelEnumerator();
+
+ this.partitionsBeingBuilt.clear();
+ double numRecordPerPartition = (double) buildRowCount / numPartitions;
+ int maxBuffer = maxInitBufferOfBucketArea(numPartitions);
+ for (int i = 0; i < numPartitions; i++) {
+ BinaryHashBucketArea area = new BinaryHashBucketArea(this, numRecordPerPartition, maxBuffer);
+ BinaryHashPartition p = new BinaryHashPartition(area, this.binaryBuildSideSerializer,
+ this.binaryProbeSideSerializer, i, recursionLevel, getNotNullNextBuffer(), this, this.segmentSize,
+ compressionEnable, compressionCodecFactory, compressionBlockSize);
+ area.setPartition(p);
+ this.partitionsBeingBuilt.add(p);
+ }
+ }
+
+ /**
+ * This method clears all partitions currently residing (partially) in memory. It releases all
+ * memory
+ * and deletes all spilled partitions.
+ *
+ * <p>This method is intended for a hard cleanup in the case that the join is aborted.
+ */
+ @Override
+ public void clearPartitions() {
+ // clear the iterators, so the next call to next() will notice
+ this.bucketIterator = null;
+ this.probeIterator = null;
+
+ for (int i = this.partitionsBeingBuilt.size() - 1; i >= 0; --i) {
+ final BinaryHashPartition p = this.partitionsBeingBuilt.get(i);
+ try {
+ p.clearAllMemory(this.availableMemory);
+ } catch (Exception e) {
+ LOG.error("Error during partition cleanup.", e);
+ }
+ }
+ this.partitionsBeingBuilt.clear();
+
+ // clear the partitions that are still to be done (that have files on disk)
+ for (final BinaryHashPartition p : this.partitionsPending) {
+ p.clearAllMemory(this.availableMemory);
+ }
+ }
+
+ /**
+ * Selects a partition and spills it. The number of the spilled partition is returned.
+ *
+ * @return The number of the spilled partition.
+ */
+ @Override
+ protected int spillPartition() throws IOException {
+ // find the largest partition
+ int largestNumBlocks = 0;
+ int largestPartNum = -1;
+
+ for (int i = 0; i < partitionsBeingBuilt.size(); i++) {
+ BinaryHashPartition p = partitionsBeingBuilt.get(i);
+ if (p.isInMemory() && p.getNumOccupiedMemorySegments() > largestNumBlocks) {
+ largestNumBlocks = p.getNumOccupiedMemorySegments();
+ largestPartNum = i;
+ }
+ }
+ final BinaryHashPartition p = partitionsBeingBuilt.get(largestPartNum);
+
+ // spill the partition
+ int numBuffersFreed = p.spillPartition(this.ioManager,
+ this.currentEnumerator.next(), this.buildSpillReturnBuffers);
+ this.buildSpillRetBufferNumbers += numBuffersFreed;
+
+ LOG.info(String.format("Grace hash join: Ran out memory, choosing partition " +
+ "[%d] to spill, %d memory segments being freed",
+ largestPartNum, numBuffersFreed));
+
+ // grab as many buffers as are available directly
+ MemorySegment currBuff;
+ while (this.buildSpillRetBufferNumbers > 0 && (currBuff = this.buildSpillReturnBuffers.poll()) != null) {
+ this.availableMemory.add(currBuff);
+ this.buildSpillRetBufferNumbers--;
+ }
+ numSpillFiles++;
+ spillInBytes += numBuffersFreed * segmentSize;
+ // The bloomFilter is built after the data is spilled, so that we can use enough memory.
+ p.buildBloomFilterAndFreeBucket();
+ return largestPartNum;
+ }
+
+ boolean applyCondition(BinaryRow candidate) throws Exception {
+ BinaryRow buildKey = buildSideProjection.apply(candidate);
+ // They come from Projection, so we can make sure it is in byte[].
+ boolean equal = buildKey.getSizeInBytes() == probeKey.getSizeInBytes()
+ && BinaryRowUtil.byteArrayEquals(
+ buildKey.getSegments()[0].getHeapMemory(),
+ probeKey.getSegments()[0].getHeapMemory(),
+ buildKey.getSizeInBytes());
+ // TODO do null filter in advance?
+ if (!nullSafe) {
+ equal = equal && !(filterAllNulls ? buildKey.anyNull() : buildKey.anyNull(nullFilterKeys));
+ }
+ return condFunc == null ? equal : equal && (reverseJoin ? condFunc.apply(probeRow, candidate)
+ : condFunc.apply(candidate, probeRow));
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryRowChannelInputViewIterator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryRowChannelInputViewIterator.java
new file mode 100644
index 0000000..bd6f1d3
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BinaryRowChannelInputViewIterator.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.typeutils.BinaryRowSerializer;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A simple iterator over the input read though an I/O channel.
+ * Use {@link BinaryRowSerializer#deserializeFromPages}.
+ */
+public class BinaryRowChannelInputViewIterator implements MutableObjectIterator<BinaryRow> {
+ private final ChannelReaderInputView inView;
+
+ private final BinaryRowSerializer serializer;
+
+ private final List<MemorySegment> freeMemTarget;
+
+ public BinaryRowChannelInputViewIterator(
+ ChannelReaderInputView inView,
+ BinaryRowSerializer serializer) {
+ this(inView, new ArrayList<>(), serializer);
+ }
+
+ public BinaryRowChannelInputViewIterator(
+ ChannelReaderInputView inView,
+ List<MemorySegment> freeMemTarget,
+ BinaryRowSerializer serializer) {
+ this.inView = inView;
+ this.freeMemTarget = freeMemTarget;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) throws IOException {
+ try {
+ return this.serializer.deserializeFromPages(reuse, this.inView);
+ } catch (EOFException eofex) {
+ final List<MemorySegment> freeMem = this.inView.close();
+ if (this.freeMemTarget != null) {
+ this.freeMemTarget.addAll(freeMem);
+ }
+ return null;
+ }
+ }
+
+ @Override
+ public BinaryRow next() throws IOException {
+ throw new UnsupportedOperationException("This method is disabled due to performance issue!");
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BuildSideIterator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BuildSideIterator.java
new file mode 100644
index 0000000..f43803b
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/BuildSideIterator.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.operators.util.BitSet;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.table.typeutils.BinaryRowSerializer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Iterate all the elements in memory which has not been(or has been) probed during probe phase.
+ */
+public class BuildSideIterator implements RowIterator<BinaryRow> {
+
+ private final BinaryRowSerializer accessor;
+
+ private final ArrayList<BinaryHashPartition> partitionsBeingBuilt;
+
+ private final BitSet probedSet;
+
+ private final boolean matchedOrUnmatched;
+
+ private int areaIndex;
+
+ private BinaryRow reuse;
+
+ private BucketIterator bucketIterator;
+
+ BuildSideIterator(
+ BinaryRowSerializer accessor,
+ BinaryRow reuse,
+ ArrayList<BinaryHashPartition> partitionsBeingBuilt,
+ BitSet probedSet,
+ boolean matchedOrUnmatched) {
+ this.accessor = accessor;
+ this.partitionsBeingBuilt = partitionsBeingBuilt;
+ this.probedSet = probedSet;
+ this.reuse = reuse;
+ this.matchedOrUnmatched = matchedOrUnmatched;
+ this.areaIndex = -1;
+ }
+
+ @Override
+ public boolean advanceNext() {
+ if (bucketIterator != null && bucketIterator.advanceNext()) {
+ return true;
+ } else {
+ areaIndex++;
+ while (areaIndex < partitionsBeingBuilt.size()) {
+ BinaryHashPartition partition = partitionsBeingBuilt.get(areaIndex);
+ if (partition.isInMemory()) {
+ BucketIterator iterator = new BucketIterator(partition.bucketArea, accessor, reuse, probedSet, matchedOrUnmatched);
+ if (iterator.advanceNext()) {
+ bucketIterator = iterator;
+ return true;
+ }
+ }
+ areaIndex++;
+ }
+ return false;
+ }
+ }
+
+ @Override
+ public BinaryRow getRow() {
+ return bucketIterator.getRow();
+ }
+
+ /**
+ * Partition bucket iterator.
+ */
+ public static class BucketIterator implements RowIterator<BinaryRow> {
+
+ private BinaryHashBucketArea area;
+
+ private final BinaryRowSerializer accessor;
+
+ private final BitSet probedSet;
+
+ private final boolean matchedOrUnmatched;
+
+ private MemorySegment bucketSegment;
+
+ private MemorySegment[] overflowSegments;
+
+ private int scanCount;
+
+ private int bucketInSegmentOffset;
+
+ private int pointerOffset;
+
+ private int countInBucket;
+
+ private int numInBucket;
+
+ private BinaryRow reuse;
+
+ private BinaryRow instance;
+
+ BucketIterator(
+ BinaryHashBucketArea area,
+ BinaryRowSerializer accessor,
+ BinaryRow reuse,
+ BitSet probedSet,
+ boolean matchedOrUnmatched) {
+ this.area = area;
+ this.accessor = accessor;
+ this.probedSet = probedSet;
+ this.reuse = reuse;
+ this.matchedOrUnmatched = matchedOrUnmatched;
+
+ scanCount = -1;
+ moveToNextBucket();
+ }
+
+ @Override
+ public boolean advanceNext() {
+ // search unprobed record in bucket, while none found move to next bucket and search.
+ while (true) {
+ this.instance = nextInBucket(reuse);
+ if (instance == null) {
+ // return null while there are no more buckets.
+ if (!moveToNextBucket()) {
+ return false;
+ }
+ } else {
+ return true;
+ }
+ }
+ }
+
+ @Override
+ public BinaryRow getRow() {
+ return instance;
+ }
+
+ /**
+ * Move to next bucket, return true while move to a on heap bucket, return false while move
+ * to a spilled bucket
+ * or there is no more bucket.
+ */
+ private boolean moveToNextBucket() {
+ scanCount++;
+ if (scanCount >= area.numBuckets) {
+ return false;
+ }
+ // move to next bucket, update all the current bucket status with new bucket information.
+ final int bucketArrayPos = scanCount >> area.table.bucketsPerSegmentBits;
+ final int currentBucketInSegmentOffset = (scanCount & area.table.bucketsPerSegmentMask) << BinaryHashBucketArea.BUCKET_SIZE_BITS;
+ MemorySegment currentBucket = area.buckets[bucketArrayPos];
+ setBucket(currentBucket, area.overflowSegments, currentBucketInSegmentOffset);
+ return true;
+ }
+
+ // update current bucket status.
+ private void setBucket(
+ MemorySegment bucket, MemorySegment[] overflowSegments,
+ int bucketInSegmentOffset) {
+ this.bucketSegment = bucket;
+ this.overflowSegments = overflowSegments;
+ this.bucketInSegmentOffset = bucketInSegmentOffset;
+ this.pointerOffset = bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_POINTER_START_OFFSET;
+ this.countInBucket = bucket.getShort(bucketInSegmentOffset + BinaryHashBucketArea.HEADER_COUNT_OFFSET);
+ this.numInBucket = 0;
+ // reset probedSet with probedFlags offset in this bucket.
+ this.probedSet.setMemorySegment(bucketSegment, this.bucketInSegmentOffset + BinaryHashBucketArea.PROBED_FLAG_OFFSET);
+ }
+
+ private BinaryRow nextInBucket(BinaryRow reuse) {
+ // loop over all segments that are involved in the bucket (original bucket plus overflow buckets)
+ while (countInBucket != 0) {
+
+ checkNotNull(bucketSegment);
+
+ while (this.numInBucket < this.countInBucket) {
+ boolean probed = probedSet.get(numInBucket);
+ numInBucket++;
+ if (matchedOrUnmatched == probed) {
+ try {
+ this.area.partition.setReadPosition(bucketSegment.getInt(pointerOffset));
+ reuse = this.accessor.mapFromPages(reuse, this.area.partition);
+ this.pointerOffset += BinaryHashBucketArea.POINTER_LEN;
+ return reuse;
+ } catch (IOException ioex) {
+ throw new RuntimeException("Error deserializing key or value from the hashtable: " +
+ ioex.getMessage(), ioex);
+ }
+ } else {
+ this.pointerOffset += BinaryHashBucketArea.POINTER_LEN;
+ }
+ }
+
+ // this segment is done. check if there is another chained bucket
+ final int forwardPointer = this.bucketSegment.getInt(this.bucketInSegmentOffset + BinaryHashBucketArea.HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BinaryHashBucketArea.BUCKET_FORWARD_POINTER_NOT_SET) {
+ return null;
+ }
+
+ final int overflowSegIndex = forwardPointer >>> area.table.segmentSizeBits;
+ this.bucketSegment = this.overflowSegments[overflowSegIndex];
+ this.bucketInSegmentOffset = forwardPointer & area.table.segmentSizeMask;
+ this.pointerOffset = bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_POINTER_START_OFFSET;
+ this.countInBucket = this.bucketSegment.getShort(this.bucketInSegmentOffset + BinaryHashBucketArea.HEADER_COUNT_OFFSET);
+ this.numInBucket = 0;
+ // reset probedSet with probedFlags offset in this bucket.
+ this.probedSet.setMemorySegment(bucketSegment, this.bucketInSegmentOffset + BinaryHashBucketArea.PROBED_FLAG_OFFSET);
+ }
+ return null;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/HashTableBloomFilter.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/HashTableBloomFilter.java
new file mode 100644
index 0000000..5297deb
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/HashTableBloomFilter.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.operators.util.BloomFilter;
+import org.apache.flink.util.MathUtils;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Bloom filters for HashTable.
+ */
+class HashTableBloomFilter {
+
+ private final MemorySegment[] buffers;
+ private final int numBuffers;
+ private final int numBuffersMask;
+ private final BloomFilter filter;
+ private final int maxSize;
+
+ private int size = 0;
+
+ HashTableBloomFilter(MemorySegment[] buffers, long numRecords) {
+ checkArgument(buffers != null && buffers.length > 0);
+ this.buffers = buffers;
+ this.numBuffers = buffers.length;
+ checkArgument(MathUtils.isPowerOf2(numBuffers));
+ this.numBuffersMask = numBuffers - 1;
+ int bufferSize = buffers[0].size();
+ this.filter = new BloomFilter((int) (numRecords / numBuffers), buffers[0].size());
+ filter.setBitsLocation(buffers[0], 0);
+
+ // We assume that a BloomFilter can contain up to 2.44 elements per byte.
+ // fpp roughly equal 0.2
+ this.maxSize = (int) ((numBuffers * bufferSize) * 2.44);
+ }
+
+ private void setLocation(int hash) {
+ if (numBuffers > 1) {
+ filter.setBitsLocation(buffers[hash & numBuffersMask], 0);
+ }
+ }
+
+ boolean testHash(int hash) {
+ setLocation(hash);
+ return filter.testHash(hash);
+ }
+
+ /**
+ * @return false if the accuracy of the BloomFilter is not high.
+ */
+ boolean addHash(int hash) {
+ setLocation(hash);
+ filter.addHash(hash);
+ size++;
+ return size <= maxSize;
+ }
+
+ public MemorySegment[] getBuffers() {
+ return buffers;
+ }
+
+ static int optimalSegmentNumber(long maxNum, int segSize, double fpp) {
+ double numBytes = optimalNumOfBits(maxNum, fpp) / 8D;
+ return (int) Math.ceil(numBytes / segSize);
+ }
+
+ private static int optimalNumOfBits(long maxNumEntries, double fpp) {
+ return (int) (-maxNumEntries * Math.log(fpp) / (Math.log(2) * Math.log(2)));
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/LookupBucketIterator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/LookupBucketIterator.java
new file mode 100644
index 0000000..1baeec8
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/LookupBucketIterator.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.util.MathUtils;
+
+/**
+ * Build iterator from bucket to match probe row.
+ */
+public class LookupBucketIterator implements RowIterator<BinaryRow> {
+
+ private final BinaryHashTable table;
+ private final int segmentSizeBits;
+ private final int segmentSizeMask;
+
+ private MemorySegment bucket;
+
+ private MemorySegment[] overflowSegments;
+
+ private BinaryHashPartition partition;
+
+ private int bucketInSegmentOffset;
+
+ private int pointerOffset;
+
+ private int searchHashCode;
+
+ private int hashCodeOffset;
+
+ private int countInBucket;
+
+ private int numInBucket;
+
+ private BinaryRow reuse;
+
+ private BinaryRow instance;
+
+ LookupBucketIterator(BinaryHashTable table) {
+ this.table = table;
+ this.reuse = table.binaryBuildSideSerializer.createInstance();
+ this.segmentSizeBits = MathUtils.log2strict(table.pageSize());
+ this.segmentSizeMask = table.pageSize() - 1;
+ }
+
+ public void set(
+ MemorySegment bucket, MemorySegment[] overflowSegments,
+ BinaryHashPartition partition,
+ int searchHashCode, int bucketInSegmentOffset) {
+ this.bucket = bucket;
+ this.overflowSegments = overflowSegments;
+ this.partition = partition;
+ this.searchHashCode = searchHashCode;
+ this.bucketInSegmentOffset = bucketInSegmentOffset;
+ this.pointerOffset = bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_POINTER_START_OFFSET;
+ this.hashCodeOffset = this.bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_HEADER_LENGTH;
+ this.countInBucket = bucket.getShort(bucketInSegmentOffset + BinaryHashBucketArea.HEADER_COUNT_OFFSET);
+ this.numInBucket = 0;
+ }
+
+ @Override
+ public boolean advanceNext() {
+ // loop over all segments that are involved in the bucket (original bucket plus overflow buckets)
+ while (countInBucket != 0) {
+ table.probedSet.setMemorySegment(bucket, this.bucketInSegmentOffset + BinaryHashBucketArea.PROBED_FLAG_OFFSET);
+ while (this.numInBucket < this.countInBucket) {
+
+ final int thisCode = this.bucket.getInt(this.hashCodeOffset);
+ this.hashCodeOffset += BinaryHashBucketArea.HASH_CODE_LEN;
+
+ this.numInBucket++;
+
+ // check if the hash code matches
+ if (thisCode == this.searchHashCode) {
+ // get the pointer to the pair
+ final int pointer = bucket.getInt(pointerOffset);
+ pointerOffset += BinaryHashBucketArea.POINTER_LEN;
+
+ // deserialize the key to check whether it is really equal, or whether we had only a hash collision
+ try {
+ this.partition.setReadPosition(pointer);
+ BinaryRow row = table.binaryBuildSideSerializer.mapFromPages(reuse, this.partition);
+ if (table.applyCondition(row)) {
+ if (table.type.needSetProbed()) {
+ table.probedSet.set(numInBucket - 1);
+ }
+ this.instance = row;
+ return true;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Error deserializing key or value from the hashtable: " +
+ e.getMessage(), e);
+ }
+ } else {
+ pointerOffset += BinaryHashBucketArea.POINTER_LEN;
+ }
+ }
+
+ // this segment is done. check if there is another chained bucket
+ final int forwardPointer = this.bucket.getInt(this.bucketInSegmentOffset + BinaryHashBucketArea.HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BinaryHashBucketArea.BUCKET_FORWARD_POINTER_NOT_SET) {
+ this.instance = null;
+ return false;
+ }
+
+ final int overflowSegIndex = forwardPointer >>> this.segmentSizeBits;
+ this.bucket = this.overflowSegments[overflowSegIndex];
+ this.bucketInSegmentOffset = forwardPointer & this.segmentSizeMask;
+ this.pointerOffset = bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_POINTER_START_OFFSET;
+ this.countInBucket = this.bucket.getShort(this.bucketInSegmentOffset + BinaryHashBucketArea.HEADER_COUNT_OFFSET);
+ this.hashCodeOffset = this.bucketInSegmentOffset + BinaryHashBucketArea.BUCKET_HEADER_LENGTH;
+ this.numInBucket = 0;
+ }
+ this.instance = null;
+ return false;
+ }
+
+ @Override
+ public BinaryRow getRow() {
+ return instance;
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/ProbeIterator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/ProbeIterator.java
new file mode 100644
index 0000000..25f2f73
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/ProbeIterator.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.BinaryRow;
+
+import java.io.IOException;
+
+/**
+ * Probe iterator from probe or spilled partition.
+ */
+public final class ProbeIterator {
+
+ private ChannelReaderInputViewIterator<BinaryRow> source;
+
+ private BaseRow instance;
+ private BinaryRow reuse;
+
+ public ProbeIterator(BinaryRow instance) {
+ this.instance = instance;
+ }
+
+ public void set(ChannelReaderInputViewIterator<BinaryRow> source) {
+ this.source = source;
+ }
+
+ public void setReuse(BinaryRow reuse) {
+ this.reuse = reuse;
+ }
+
+ public BinaryRow next() throws IOException {
+ BinaryRow retVal = this.source.next(reuse);
+ if (retVal != null) {
+ this.instance = retVal;
+ return retVal;
+ } else {
+ return null;
+ }
+ }
+
+ public BaseRow current() {
+ return this.instance;
+ }
+
+ public void setInstance(BaseRow instance) {
+ this.instance = instance;
+ }
+
+ public boolean hasSource() {
+ return source != null;
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/WrappedRowIterator.java
similarity index 51%
copy from flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java
copy to flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/WrappedRowIterator.java
index c84db7c..acd9ccf 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/hashtable/WrappedRowIterator.java
@@ -16,40 +16,40 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.sort;
+package org.apache.flink.table.runtime.hashtable;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
-import org.apache.flink.table.runtime.util.MemorySegmentPool;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.util.MutableObjectIterator;
-import java.util.List;
+import java.io.IOException;
/**
- * Test pool.
+ * Wrap {@link MutableObjectIterator} to java {@link RowIterator}.
*/
-public class TestMemorySegmentPool implements MemorySegmentPool {
+public class WrappedRowIterator<T extends BaseRow> implements RowIterator<T> {
- private int pageSize;
+ private final MutableObjectIterator<T> iterator;
+ private final T reuse;
+ private T instance;
- public TestMemorySegmentPool(int pageSize) {
- this.pageSize = pageSize;
+ public WrappedRowIterator(MutableObjectIterator<T> iterator, T reuse) {
+ this.iterator = iterator;
+ this.reuse = reuse;
}
@Override
- public int pageSize() {
- return pageSize;
+ public boolean advanceNext() {
+ try {
+ this.instance = iterator.next(reuse);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return this.instance != null;
}
@Override
- public void returnAll(List<MemorySegment> memory) {
- }
-
- @Override
- public void clear() {
- }
-
- @Override
- public MemorySegment nextSegment() {
- return MemorySegmentFactory.wrap(new byte[pageSize]);
+ public T getRow() {
+ return instance;
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/HashJoinType.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/HashJoinType.java
new file mode 100644
index 0000000..a60ccdf
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/HashJoinType.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join;
+
+/**
+ * Join type for hash table.
+ */
+public enum HashJoinType {
+
+ INNER,
+ BUILD_OUTER,
+ PROBE_OUTER,
+ FULL_OUTER,
+ SEMI,
+ ANTI,
+ BUILD_LEFT_SEMI,
+ BUILD_LEFT_ANTI;
+
+ public boolean isBuildOuter() {
+ return this.equals(BUILD_OUTER) || this.equals(FULL_OUTER);
+ }
+
+ public boolean isProbeOuter() {
+ return this.equals(PROBE_OUTER) || this.equals(FULL_OUTER);
+ }
+
+ public boolean buildLeftSemiOrAnti() {
+ return this.equals(BUILD_LEFT_SEMI) || this.equals(BUILD_LEFT_ANTI);
+ }
+
+ public boolean needSetProbed() {
+ return isBuildOuter() || buildLeftSemiOrAnti();
+ }
+
+ public static HashJoinType of(boolean leftIsBuild, boolean leftOuter, boolean rightOuter) {
+ if (leftOuter && rightOuter) {
+ return FULL_OUTER;
+ } else if (leftOuter) {
+ return leftIsBuild ? BUILD_OUTER : PROBE_OUTER;
+ } else if (rightOuter) {
+ return leftIsBuild ? PROBE_OUTER : BUILD_OUTER;
+ } else {
+ return INNER;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/NullAwareJoinHelper.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/NullAwareJoinHelper.java
new file mode 100644
index 0000000..c0dcb6f
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/NullAwareJoinHelper.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.join;
+
+import org.apache.flink.table.dataformat.BinaryRow;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Helper for null aware join.
+ */
+public class NullAwareJoinHelper {
+
+ public static int[] getNullFilterKeys(boolean[] filterNulls) {
+ checkArgument(filterNulls.length > 0);
+ List<Integer> nullFilterKeyList = new ArrayList<>();
+ for (int i = 0; i < filterNulls.length; i++) {
+ if (filterNulls[i]) {
+ nullFilterKeyList.add(i);
+ }
+ }
+ return ArrayUtils.toPrimitive(nullFilterKeyList.toArray(new Integer[0]));
+ }
+
+ public static boolean shouldFilter(
+ boolean nullSafe, boolean filterAllNulls,
+ int[] nullFilterKeys, BinaryRow key) {
+ return !nullSafe && (filterAllNulls ? key.anyNull() : key.anyNull(nullFilterKeys));
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryIndexedSortable.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryIndexedSortable.java
index a4606c1..a6e13f0 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryIndexedSortable.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryIndexedSortable.java
@@ -241,7 +241,6 @@ public abstract class BinaryIndexedSortable implements IndexedSortable {
// go through all records in the memory segment
for (int offset = 0; currentRecord < numRecords && offset <= this.lastIndexEntryOffset; currentRecord++, offset += this.indexEntrySize) {
final long pointer = currentIndexSegment.getLong(offset);
-
this.recordBuffer.setReadPosition(pointer);
this.serializer.copyFromPagesToView(this.recordBuffer, output);
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryKVInMemorySortBuffer.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryKVInMemorySortBuffer.java
index 642d81b..d4ca5b2 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryKVInMemorySortBuffer.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/BinaryKVInMemorySortBuffer.java
@@ -82,8 +82,7 @@ public class BinaryKVInMemorySortBuffer extends BinaryIndexedSortable {
}
}
- private void load(long numElements,
- RandomAccessInputView recordInputView) throws IOException {
+ private void load(long numElements, RandomAccessInputView recordInputView) throws IOException {
for (int index = 0; index < numElements; index++) {
serializer.checkSkipReadForFixLengthPart(recordInputView);
long pointer = recordInputView.getReadPosition();
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/ListMemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/ListMemorySegmentPool.java
index 3523997..ef4a128 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/ListMemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/sort/ListMemorySegmentPool.java
@@ -55,7 +55,6 @@ public class ListMemorySegmentPool implements MemorySegmentPool {
segments.addAll(memory);
}
- @Override
public void clear() {
segments.clear();
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
index 94713e5..a094714 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/MemorySegmentPool.java
@@ -21,25 +21,26 @@ package org.apache.flink.table.runtime.util;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentSource;
-import java.util.Arrays;
import java.util.List;
/**
- * MemorySegment pool.
+ * MemorySegment pool to hold pages in memory.
*/
public interface MemorySegmentPool extends MemorySegmentSource {
+ /**
+ * Get the page size of each page this pool holds.
+ *
+ * @return the page size
+ */
int pageSize();
+ /**
+ * Return all pages back into this pool.
+ *
+ * @param memory the pages which want to be returned.
+ */
void returnAll(List<MemorySegment> memory);
- default void returnAll(MemorySegment[] memory) {
- returnAll(Arrays.asList(memory));
- }
-
- void clear();
-
- default int remainBuffers() {
- throw new UnsupportedOperationException();
- }
}
+
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/RowIterator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/RowIterator.java
new file mode 100644
index 0000000..4e5a91e
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/RowIterator.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.util;
+
+import org.apache.flink.table.dataformat.BaseRow;
+
+import java.util.Iterator;
+
+/**
+ * An internal iterator interface which presents a more restrictive API than {@link Iterator}.
+ *
+ * <p>One major departure from the Java iterator API is the fusing of the `hasNext()` and `next()`
+ * calls: Java's iterator allows users to call `hasNext()` without immediately advancing the
+ * iterator to consume the next row, whereas {@link RowIterator} combines these calls into a single
+ * {@link #advanceNext()} method.
+ */
+public interface RowIterator<T extends BaseRow> {
+ /**
+ * Advance this iterator by a single row. Returns `false` if this iterator has no more rows
+ * and `true` otherwise. If this returns `true`, then the new row can be retrieved by calling
+ * {@link #getRow()}.
+ */
+ boolean advanceNext();
+
+ /**
+ * Retrieve the row from this iterator. This method is idempotent. It is illegal to call this
+ * method after [[advanceNext()]] has returned `false`.
+ */
+ T getRow();
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/AbstractRowSerializer.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/AbstractRowSerializer.java
index 4fe561b..4d99b95 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/AbstractRowSerializer.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/AbstractRowSerializer.java
@@ -32,64 +32,75 @@ import java.io.IOException;
public abstract class AbstractRowSerializer<T extends BaseRow> extends TypeSerializer<T> {
/**
+ * Get the number of fields.
+ */
+ public abstract int getArity();
+
+ /**
* Convert a {@link BaseRow} to a {@link BinaryRow}.
*/
public abstract BinaryRow baseRowToBinary(T baseRow) throws IOException;
/**
- * Serializes the given record to the given target paged output view. Make specific implementers decide whether
- * or not to jump, how many positions jump. If jumps, will skip some offset.
+ * Serializes the given record to the given target paged output view. Some implementations may
+ * skip some bytes if current page does not have enough space left, .e.g {@link BinaryRow}.
*
* @param record The record to serialize.
* @param target The output view to write the serialized data to.
- * @return Returns how much offset is skipped.
- * @throws IOException Thrown, if the serialization encountered an I/O related error. Typically raised by the
- * output view, which may have an underlying I/O channel to which it delegates.
+ *
+ * @return Returns the skipped number of bytes.
+ *
+ * @throws IOException Thrown, if the serialization encountered an I/O related error.
+ * Typically raised by the output view, which may have an underlying
+ * I/O channel to which it delegates.
*/
- public int serializeToPages(T record, AbstractPagedOutputView target) throws IOException {
- serialize(record, target);
- return 0;
- }
+ public abstract int serializeToPages(T record, AbstractPagedOutputView target)
+ throws IOException;
/**
- * De-serializes a record from the given source paged input view. Make specific implementers decide whether
- * or not to jump, how many positions jump. If jumps, will skip some bytes to read.
+ * De-serializes a record from the given source paged input view. For consistency with serialize
+ * format, some implementations may need to skip some bytes of source before de-serializing,
+ * .e.g {@link BinaryRow}. Typically, the content read from source should be copied out when
+ * de-serializing, and we are not expecting the underlying data from source is reused. If you
+ * have such requirement, see {@link #mapFromPages(AbstractPagedInputView)}.
*
* @param source The input view from which to read the data.
- * @return The deserialized element.
*
- * @throws IOException Thrown, if the de-serialization encountered an I/O related error. Typically raised by the
- * input view, which may have an underlying I/O channel from which it reads.
+ * @return The de-serialized element.
+ *
+ * @throws IOException Thrown, if the de-serialization encountered an I/O related error.
+ * Typically raised by the input view, which may have an underlying I/O
+ * channel from which it reads.
*/
- public T deserializeFromPages(AbstractPagedInputView source) throws IOException {
- return deserialize(source);
- }
+ public abstract T deserializeFromPages(AbstractPagedInputView source) throws IOException;
/**
* Reuse version of {@link #deserializeFromPages(AbstractPagedInputView)}.
*/
- public T deserializeFromPages(T reuse, AbstractPagedInputView source) throws IOException {
- return deserialize(reuse, source);
- }
+ public abstract T deserializeFromPages(T reuse, AbstractPagedInputView source)
+ throws IOException;
/**
- * Map a record from the given source paged input view. This method provides a possibility to achieve zero copy,
- * you can copy, you can not copy, while you have to properly manage the life cycle of the page.
- * Pay attention: Before the end of the visit(Or iterator), pages can not be released.
+ * Map a record from the given source paged input view. This method provides a possibility to
+ * achieve zero copy when de-serializing. You can either choose copy or not copy the content
+ * read from source, but we encourage to make it zero copy.
+ *
+ * <p>If you choose the zero copy way, you have to deal with the lifecycle of the pages
+ * properly.
+ *
* @param source The input view from which to read the data.
- * @return The deserialized element.
*
- * @throws IOException Thrown, if the de-serialization encountered an I/O related error. Typically raised by the
- * input view, which may have an underlying I/O channel from which it reads.
+ * @return The mapped element.
+ *
+ * @throws IOException Thrown, if the de-serialization encountered an I/O related error.
+ * Typically raised by the input view, which may have an underlying I/O
+ * channel from which it reads.
*/
- public T mapFromPages(AbstractPagedInputView source) throws IOException {
- return deserialize(source);
- }
+ public abstract T mapFromPages(AbstractPagedInputView source) throws IOException;
/**
* Reuse version of {@link #mapFromPages(AbstractPagedInputView)}.
*/
- public T mapFromPages(T reuse, AbstractPagedInputView source) throws IOException {
- return deserialize(reuse, source);
- }
+ public abstract T mapFromPages(T reuse, AbstractPagedInputView source) throws IOException;
+
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
index d4f8246..e5e30ad 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
@@ -65,10 +65,6 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
this.binarySerializer = new BinaryRowSerializer(types.length);
}
- public int getArity() {
- return types.length;
- }
-
@Override
public TypeSerializer<BaseRow> duplicate() {
return new BaseRowSerializer(types, fieldSerializers);
@@ -81,6 +77,25 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
}
@Override
+ public void serialize(BaseRow row, DataOutputView target) throws IOException {
+ binarySerializer.serialize(baseRowToBinary(row), target);
+ }
+
+ @Override
+ public BaseRow deserialize(DataInputView source) throws IOException {
+ return binarySerializer.deserialize(source);
+ }
+
+ @Override
+ public BaseRow deserialize(BaseRow reuse, DataInputView source) throws IOException {
+ if (reuse instanceof BinaryRow) {
+ return binarySerializer.deserialize((BinaryRow) reuse, source);
+ } else {
+ return binarySerializer.deserialize(source);
+ }
+ }
+
+ @Override
public BaseRow copy(BaseRow from) {
if (from.getArity() != types.length) {
throw new IllegalArgumentException("Row arity: " + from.getArity() +
@@ -101,9 +116,9 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
", but serializer arity: " + types.length);
}
if (from instanceof BinaryRow) {
- return reuse instanceof BinaryRow ?
- ((BinaryRow) from).copy((BinaryRow) reuse) :
- ((BinaryRow) from).copy();
+ return reuse instanceof BinaryRow
+ ? ((BinaryRow) from).copy((BinaryRow) reuse)
+ : ((BinaryRow) from).copy();
} else {
return copyBaseRow(from, reuse);
}
@@ -120,7 +135,10 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
ret.setHeader(from.getHeader());
for (int i = 0; i < from.getArity(); i++) {
if (!from.isNullAt(i)) {
- ret.setField(i, fieldSerializers[i].copy((TypeGetterSetters.get(from, i, types[i]))));
+ ret.setField(
+ i,
+ fieldSerializers[i].copy((TypeGetterSetters.get(from, i, types[i])))
+ );
} else {
ret.setNullAt(i);
}
@@ -133,6 +151,11 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
binarySerializer.copy(source, target);
}
+ @Override
+ public int getArity() {
+ return types.length;
+ }
+
/**
* Convert base row to binary row.
* TODO modify it to code gen, and reuse BinaryRow&BinaryRowWriter.
@@ -156,25 +179,6 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
}
@Override
- public void serialize(BaseRow row, DataOutputView target) throws IOException {
- binarySerializer.serialize(baseRowToBinary(row), target);
- }
-
- @Override
- public BaseRow deserialize(DataInputView source) throws IOException {
- return binarySerializer.deserialize(source);
- }
-
- @Override
- public BaseRow deserialize(BaseRow reuse, DataInputView source) throws IOException {
- if (reuse instanceof BinaryRow) {
- return binarySerializer.deserialize((BinaryRow) reuse, source);
- } else {
- return binarySerializer.deserialize(source);
- }
- }
-
- @Override
public int serializeToPages(BaseRow row, AbstractPagedOutputView target) throws IOException {
return binarySerializer.serializeToPages(baseRowToBinary(row), target);
}
@@ -185,7 +189,8 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
}
@Override
- public BaseRow deserializeFromPages(BaseRow reuse,
+ public BaseRow deserializeFromPages(
+ BaseRow reuse,
AbstractPagedInputView source) throws IOException {
throw new UnsupportedOperationException("Not support!");
}
@@ -197,7 +202,8 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
}
@Override
- public BaseRow mapFromPages(BaseRow reuse,
+ public BaseRow mapFromPages(
+ BaseRow reuse,
AbstractPagedInputView source) throws IOException {
if (reuse instanceof BinaryRow) {
return binarySerializer.mapFromPages((BinaryRow) reuse, source);
@@ -252,7 +258,8 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
BaseRowSerializerSnapshot(InternalType[] types, TypeSerializer[] serializers) {
this.previousTypes = types;
- this.nestedSerializersSnapshotDelegate = new NestedSerializersSnapshotDelegate(serializers);
+ this.nestedSerializersSnapshotDelegate = new NestedSerializersSnapshotDelegate(
+ serializers);
}
@Override
@@ -271,23 +278,34 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
}
@Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+ throws IOException {
int length = in.readInt();
DataInputViewStream stream = new DataInputViewStream(in);
previousTypes = new InternalType[length];
for (int i = 0; i < length; i++) {
try {
- previousTypes[i] = InstantiationUtil.deserializeObject(stream, userCodeClassLoader);
- } catch (ClassNotFoundException e) {
+ previousTypes[i] = InstantiationUtil.deserializeObject(
+ stream,
+ userCodeClassLoader
+ );
+ }
+ catch (ClassNotFoundException e) {
throw new IOException(e);
}
}
- this.nestedSerializersSnapshotDelegate = NestedSerializersSnapshotDelegate.readNestedSerializerSnapshots(in, userCodeClassLoader);
+ this.nestedSerializersSnapshotDelegate = NestedSerializersSnapshotDelegate.readNestedSerializerSnapshots(
+ in,
+ userCodeClassLoader
+ );
}
@Override
public BaseRowSerializer restoreSerializer() {
- return new BaseRowSerializer(previousTypes, nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
+ return new BaseRowSerializer(
+ previousTypes,
+ nestedSerializersSnapshotDelegate.getRestoredNestedSerializers()
+ );
}
@Override
@@ -304,11 +322,13 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<BaseRow> intermediateResult =
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
newRowSerializer.fieldSerializers,
- nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots());
+ nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots()
+ );
if (intermediateResult.isCompatibleWithReconfiguredSerializer()) {
BaseRowSerializer reconfiguredCompositeSerializer = restoreSerializer();
- return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(reconfiguredCompositeSerializer);
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
+ reconfiguredCompositeSerializer);
}
return intermediateResult.getFinalResult();
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
index 7e11ed3..fa2828b 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
@@ -70,19 +70,6 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
return copy(from, new BinaryRow(numFields));
}
- /**
- * Copy a page binaryRow to output view.
- * @param source source paged input view.
- * @param target output view.
- */
- public void copyFromPagesToView(
- AbstractPagedInputView source, DataOutputView target) throws IOException {
- checkSkipReadForFixLengthPart(source);
- int length = source.readInt();
- target.writeInt(length);
- target.write(source, length);
- }
-
@Override
public BinaryRow copy(BinaryRow from, BinaryRow reuse) {
return from.copy(reuse);
@@ -94,14 +81,14 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
@Override
- public BinaryRow baseRowToBinary(BinaryRow baseRow) throws IOException {
- return baseRow;
- }
-
- @Override
public void serialize(BinaryRow record, DataOutputView target) throws IOException {
target.writeInt(record.getSizeInBytes());
- SegmentsUtil.copyBytesToView(record.getSegments(), record.getOffset(), record.getSizeInBytes(), target);
+ SegmentsUtil.copyBytesToView(
+ record.getSegments(),
+ record.getOffset(),
+ record.getSizeInBytes(),
+ target
+ );
}
@Override
@@ -122,17 +109,29 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
int length = source.readInt();
if (segments == null || segments[0].size() < length) {
- segments = new MemorySegment[] {MemorySegmentFactory.wrap(new byte[length])};
+ segments = new MemorySegment[]{MemorySegmentFactory.wrap(new byte[length])};
}
source.readFully(segments[0].getArray(), 0, length);
reuse.pointTo(segments, 0, length);
return reuse;
}
- // ============================ Pages serializers ===================================
+ @Override
+ public int getArity() {
+ return numFields;
+ }
+
+ @Override
+ public BinaryRow baseRowToBinary(BinaryRow baseRow) throws IOException {
+ return baseRow;
+ }
+
+ // ============================ Page related operations ===================================
@Override
- public int serializeToPages(BinaryRow record, AbstractPagedOutputView headerLessView) throws IOException {
+ public int serializeToPages(
+ BinaryRow record,
+ AbstractPagedOutputView headerLessView) throws IOException {
checkArgument(headerLessView.getHeaderLength() == 0);
int skip = checkSkipWriteForFixLengthPart(headerLessView);
serialize(record, headerLessView);
@@ -140,12 +139,15 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
@Override
- public BinaryRow deserializeFromPages(AbstractPagedInputView headerLessView) throws IOException {
+ public BinaryRow deserializeFromPages(
+ AbstractPagedInputView headerLessView) throws IOException {
return deserializeFromPages(createInstance(), headerLessView);
}
@Override
- public BinaryRow deserializeFromPages(BinaryRow reuse, AbstractPagedInputView headerLessView) throws IOException {
+ public BinaryRow deserializeFromPages(
+ BinaryRow reuse,
+ AbstractPagedInputView headerLessView) throws IOException {
checkArgument(headerLessView.getHeaderLength() == 0);
checkSkipReadForFixLengthPart(headerLessView);
return deserialize(reuse, headerLessView);
@@ -157,7 +159,9 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
@Override
- public BinaryRow mapFromPages(BinaryRow reuse, AbstractPagedInputView headerLessView) throws IOException {
+ public BinaryRow mapFromPages(
+ BinaryRow reuse,
+ AbstractPagedInputView headerLessView) throws IOException {
checkArgument(headerLessView.getHeaderLength() == 0);
checkSkipReadForFixLengthPart(headerLessView);
pointTo(headerLessView.readInt(), reuse, headerLessView);
@@ -165,21 +169,39 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
/**
+ * Copy a binaryRow which stored in paged input view to output view.
+ *
+ * @param source source paged input view where the binary row stored
+ * @param target the target output view.
+ */
+ public void copyFromPagesToView(
+ AbstractPagedInputView source, DataOutputView target) throws IOException {
+ checkSkipReadForFixLengthPart(source);
+ int length = source.readInt();
+ target.writeInt(length);
+ target.write(source, length);
+ }
+
+ /**
* Point row to memory segments with offset(in the AbstractPagedInputView) and length.
*
- * @param length row length.
- * @param reuse reuse BinaryRow object.
+ * @param length row length.
+ * @param reuse reuse BinaryRow object.
* @param headerLessView source memory segments container.
*/
- public void pointTo(int length, BinaryRow reuse, AbstractPagedInputView headerLessView) throws IOException {
+ public void pointTo(int length, BinaryRow reuse, AbstractPagedInputView headerLessView)
+ throws IOException {
checkArgument(headerLessView.getHeaderLength() == 0);
if (length < 0) {
throw new IOException(String.format(
"Read unexpected bytes in source of positionInSegment[%d] and limitInSegment[%d]",
- headerLessView.getCurrentPositionInSegment(), headerLessView.getCurrentSegmentLimit()));
+ headerLessView.getCurrentPositionInSegment(),
+ headerLessView.getCurrentSegmentLimit()
+ ));
}
- int remainInSegment = headerLessView.getCurrentSegmentLimit() - headerLessView.getCurrentPositionInSegment();
+ int remainInSegment = headerLessView.getCurrentSegmentLimit()
+ - headerLessView.getCurrentPositionInSegment();
MemorySegment currSeg = headerLessView.getCurrentSegment();
int currPosInSeg = headerLessView.getCurrentPositionInSegment();
if (remainInSegment >= length) {
@@ -187,7 +209,14 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
reuse.pointTo(currSeg, currPosInSeg, length);
headerLessView.skipBytesToRead(length);
} else {
- pointToMultiSegments(reuse, headerLessView, length, length - remainInSegment, currSeg, currPosInSeg);
+ pointToMultiSegments(
+ reuse,
+ headerLessView,
+ length,
+ length - remainInSegment,
+ currSeg,
+ currPosInSeg
+ );
}
}
@@ -232,8 +261,7 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
/**
* We need skip bytes to read when the remain bytes of current segment is not
- * enough to write binary row fixed part.
- * See {@link BinaryRow}.
+ * enough to write binary row fixed part. See {@link BinaryRow}.
*/
public void checkSkipReadForFixLengthPart(AbstractPagedInputView source) throws IOException {
// skip if there is no enough size.
@@ -248,7 +276,7 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
* Return fixed part length to serialize one row.
*/
public int getSerializedRowFixedPartLength() {
- return getFixedLengthPartSize() + 4; // ADD 4 size length.
+ return getFixedLengthPartSize() + LENGTH_SIZE_IN_BYTES;
}
public int getFixedLengthPartSize() {
@@ -281,7 +309,8 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
/**
* {@link TypeSerializerSnapshot} for {@link BinaryRowSerializer}.
*/
- public static final class BinaryRowSerializerSnapshot implements TypeSerializerSnapshot<BinaryRow> {
+ public static final class BinaryRowSerializerSnapshot
+ implements TypeSerializerSnapshot<BinaryRow> {
private static final int CURRENT_VERSION = 3;
private int previousNumFields;
@@ -306,7 +335,8 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
@Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+ throws IOException {
this.previousNumFields = in.readInt();
}
@@ -316,7 +346,8 @@ public class BinaryRowSerializer extends AbstractRowSerializer<BinaryRow> {
}
@Override
- public TypeSerializerSchemaCompatibility<BinaryRow> resolveSchemaCompatibility(TypeSerializer<BinaryRow> newSerializer) {
+ public TypeSerializerSchemaCompatibility<BinaryRow> resolveSchemaCompatibility(
+ TypeSerializer<BinaryRow> newSerializer) {
if (!(newSerializer instanceof BinaryRowSerializer)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/hashtable/BinaryHashTableTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/hashtable/BinaryHashTableTest.java
new file mode 100644
index 0000000..ee08da5
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/hashtable/BinaryHashTableTest.java
@@ -0,0 +1,933 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.hashtable;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemoryType;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.testutils.UnionIterator;
+import org.apache.flink.table.api.TableConfigOptions;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.BinaryRowWriter;
+import org.apache.flink.table.generated.Projection;
+import org.apache.flink.table.runtime.join.HashJoinType;
+import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.table.runtime.util.UniformBinaryRowGenerator;
+import org.apache.flink.table.typeutils.BinaryRowSerializer;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Hash table it case for binary row.
+ */
+@RunWith(Parameterized.class)
+public class BinaryHashTableTest {
+
+ private static final int PAGE_SIZE = 32 * 1024;
+ private IOManager ioManager;
+ private BinaryRowSerializer buildSideSerializer;
+ private BinaryRowSerializer probeSideSerializer;
+
+ private boolean useCompress;
+ private Configuration conf;
+
+ public BinaryHashTableTest(boolean useCompress) {
+ this.useCompress = useCompress;
+ }
+
+ @Parameterized.Parameters(name = "useCompress-{0}")
+ public static List<Boolean> getVarSeg() {
+ return Arrays.asList(true, false);
+ }
+
+ @Before
+ public void setup() {
+ TypeInformation[] types = new TypeInformation[]{Types.INT, Types.INT};
+ this.buildSideSerializer = new BinaryRowSerializer(types.length);
+ this.probeSideSerializer = new BinaryRowSerializer(types.length);
+
+ this.ioManager = new IOManagerAsync();
+
+ conf = new Configuration();
+ conf.setBoolean(TableConfigOptions.SQL_EXEC_SPILL_COMPRESSION_ENABLED, useCompress);
+ }
+
+ @After
+ public void tearDown() {
+ // shut down I/O manager and Memory Manager and verify the correct shutdown
+ this.ioManager.shutdown();
+ if (!this.ioManager.isProperlyShutDown()) {
+ fail("I/O manager was not property shut down.");
+ }
+ }
+
+ @Test
+ public void testIOBufferCountComputation() {
+ assertEquals(1, BinaryHashTable.getNumWriteBehindBuffers(32));
+ assertEquals(1, BinaryHashTable.getNumWriteBehindBuffers(33));
+ assertEquals(1, BinaryHashTable.getNumWriteBehindBuffers(40));
+ assertEquals(1, BinaryHashTable.getNumWriteBehindBuffers(64));
+ assertEquals(1, BinaryHashTable.getNumWriteBehindBuffers(127));
+ assertEquals(2, BinaryHashTable.getNumWriteBehindBuffers(128));
+ assertEquals(2, BinaryHashTable.getNumWriteBehindBuffers(129));
+ assertEquals(2, BinaryHashTable.getNumWriteBehindBuffers(511));
+ assertEquals(3, BinaryHashTable.getNumWriteBehindBuffers(512));
+ assertEquals(3, BinaryHashTable.getNumWriteBehindBuffers(513));
+ assertEquals(3, BinaryHashTable.getNumWriteBehindBuffers(2047));
+ assertEquals(4, BinaryHashTable.getNumWriteBehindBuffers(2048));
+ assertEquals(4, BinaryHashTable.getNumWriteBehindBuffers(2049));
+ assertEquals(4, BinaryHashTable.getNumWriteBehindBuffers(8191));
+ assertEquals(5, BinaryHashTable.getNumWriteBehindBuffers(8192));
+ assertEquals(5, BinaryHashTable.getNumWriteBehindBuffers(8193));
+ assertEquals(5, BinaryHashTable.getNumWriteBehindBuffers(32767));
+ assertEquals(6, BinaryHashTable.getNumWriteBehindBuffers(32768));
+ assertEquals(6, BinaryHashTable.getNumWriteBehindBuffers(Integer.MAX_VALUE));
+ }
+
+ @Test
+ public void testInMemoryMutableHashTable() throws IOException {
+ final int numKeys = 100000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+ MemoryManager memManager = new MemoryManager(896 * PAGE_SIZE, 1);
+ // ----------------------------------------------------------------------------------------
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 100 * PAGE_SIZE, ioManager);
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput);
+ Assert.assertEquals("Wrong number of records in join result.", numKeys * buildValsPerKey * probeValsPerKey, numRecordsInJoinResult);
+
+ table.close();
+
+ table.free();
+ }
+
+ private int join(
+ BinaryHashTable table,
+ MutableObjectIterator<BinaryRow> buildInput,
+ MutableObjectIterator<BinaryRow> probeInput) throws IOException {
+ return join(table, buildInput, probeInput, false);
+ }
+
+ private int join(
+ BinaryHashTable table,
+ MutableObjectIterator<BinaryRow> buildInput,
+ MutableObjectIterator<BinaryRow> probeInput,
+ boolean buildOuterJoin) throws IOException {
+ int count = 0;
+
+ BinaryRow reuseBuildSizeRow = buildSideSerializer.createInstance();
+ BinaryRow buildRow;
+ while ((buildRow = buildInput.next(reuseBuildSizeRow)) != null) {
+ table.putBuildRow(buildRow);
+ }
+ table.endBuild();
+
+ BinaryRow probeRow = probeSideSerializer.createInstance();
+ while ((probeRow = probeInput.next(probeRow)) != null) {
+ if (table.tryProbe(probeRow)){
+ count += joinWithNextKey(table, buildOuterJoin);
+ }
+ }
+
+ while (table.nextMatching()){
+ count += joinWithNextKey(table, buildOuterJoin);
+ }
+ return count;
+ }
+
+ private int joinWithNextKey(BinaryHashTable table, boolean buildOuterJoin) throws IOException {
+ int count = 0;
+ final RowIterator<BinaryRow> buildIterator = table.getBuildSideIterator();
+ final BaseRow probeRow = table.getCurrentProbeRow();
+ BinaryRow buildRow;
+
+ buildRow = buildIterator.advanceNext() ? buildIterator.getRow() : null;
+ // get the first build side value
+ if (probeRow != null && buildRow != null) {
+ count++;
+ while (buildIterator.advanceNext()) {
+ count++;
+ }
+ } else {
+ if (buildOuterJoin && probeRow == null && buildRow != null) {
+ count++;
+ while (buildIterator.advanceNext()) {
+ count++;
+ }
+ }
+ }
+ return count;
+ }
+
+ @Test
+ public void testSpillingHashJoinOneRecursionPerformance() throws IOException {
+ final int numKeys = 1000000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ // allocate the memory for the HashTable
+ MemoryManager memManager = new MemoryManager(200 * PAGE_SIZE, 1, PAGE_SIZE, MemoryType.HEAP, true);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer,
+ this.probeSideSerializer,
+ new MyProjection(),
+ new MyProjection(),
+ memManager,
+ 100 * PAGE_SIZE,
+ ioManager);
+
+ // ----------------------------------------------------------------------------------------
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput);
+
+ Assert.assertEquals("Wrong number of records in join result.", numKeys * buildValsPerKey * probeValsPerKey, numRecordsInJoinResult);
+
+ table.close();
+
+ // ----------------------------------------------------------------------------------------
+
+ table.free();
+ }
+
+ @Test
+ public void testSpillingHashJoinOneRecursionValidity() throws IOException {
+ final int numKeys = 1000000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ // create the map for validating the results
+ HashMap<Integer, Long> map = new HashMap<>(numKeys);
+
+ // ----------------------------------------------------------------------------------------
+ MemoryManager memManager = new MemoryManager(896 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 100 * PAGE_SIZE, ioManager);
+ final BinaryRow recordReuse = new BinaryRow(2);
+
+ BinaryRow buildRow = buildSideSerializer.createInstance();
+ while ((buildRow = buildInput.next(buildRow)) != null) {
+ table.putBuildRow(buildRow);
+ }
+ table.endBuild();
+
+ BinaryRow probeRow = probeSideSerializer.createInstance();
+ while ((probeRow = probeInput.next(probeRow)) != null) {
+ if (table.tryProbe(probeRow)){
+ testJoin(table, map);
+ }
+ }
+
+ while (table.nextMatching()){
+ testJoin(table, map);
+ }
+
+ table.close();
+
+ Assert.assertEquals("Wrong number of keys", numKeys, map.size());
+ for (Map.Entry<Integer, Long> entry : map.entrySet()) {
+ long val = entry.getValue();
+ int key = entry.getKey();
+
+ Assert.assertEquals("Wrong number of values in per-key cross product for key " + key,
+ probeValsPerKey * buildValsPerKey, val);
+ }
+
+ // ----------------------------------------------------------------------------------------
+
+ table.free();
+ }
+
+ @Test
+ public void testSpillingHashJoinWithMassiveCollisions() throws IOException {
+ // the following two values are known to have a hash-code collision on the initial level.
+ // we use them to make sure one partition grows over-proportionally large
+ final int repeatedValue1 = 40559;
+ final int repeatedValue2 = 92882;
+ final int repeatedValueCountBuild = 200000;
+ final int repeatedValueCountProbe = 5;
+
+ final int numKeys = 1000000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key, plus 400k pairs with two colliding keys
+ MutableObjectIterator<BinaryRow> build1 = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+ MutableObjectIterator<BinaryRow> build2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, repeatedValueCountBuild);
+ MutableObjectIterator<BinaryRow> build3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, repeatedValueCountBuild);
+ List<MutableObjectIterator<BinaryRow>> builds = new ArrayList<>();
+ builds.add(build1);
+ builds.add(build2);
+ builds.add(build3);
+ MutableObjectIterator<BinaryRow> buildInput = new UnionIterator<>(builds);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probe1 = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+ MutableObjectIterator<BinaryRow> probe2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, 5);
+ MutableObjectIterator<BinaryRow> probe3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, 5);
+ List<MutableObjectIterator<BinaryRow>> probes = new ArrayList<>();
+ probes.add(probe1);
+ probes.add(probe2);
+ probes.add(probe3);
+ MutableObjectIterator<BinaryRow> probeInput = new UnionIterator<>(probes);
+
+ // create the map for validating the results
+ HashMap<Integer, Long> map = new HashMap<>(numKeys);
+ MemoryManager memManager = new MemoryManager(896 * PAGE_SIZE, 1);
+ // ----------------------------------------------------------------------------------------
+
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 896 * PAGE_SIZE, ioManager);
+
+ final BinaryRow recordReuse = new BinaryRow(2);
+
+ BinaryRow buildRow = buildSideSerializer.createInstance();
+ while ((buildRow = buildInput.next(buildRow)) != null) {
+ table.putBuildRow(buildRow);
+ }
+ table.endBuild();
+
+ BinaryRow probeRow = probeSideSerializer.createInstance();
+ while ((probeRow = probeInput.next(probeRow)) != null) {
+ if (table.tryProbe(probeRow)){
+ testJoin(table, map);
+ }
+ }
+
+ while (table.nextMatching()){
+ testJoin(table, map);
+ }
+
+ table.close();
+
+ Assert.assertEquals("Wrong number of keys", numKeys, map.size());
+ for (Map.Entry<Integer, Long> entry : map.entrySet()) {
+ long val = entry.getValue();
+ int key = entry.getKey();
+
+ Assert.assertEquals("Wrong number of values in per-key cross product for key " + key,
+ (key == repeatedValue1 || key == repeatedValue2) ?
+ (probeValsPerKey + repeatedValueCountProbe) * (buildValsPerKey + repeatedValueCountBuild) :
+ probeValsPerKey * buildValsPerKey, val);
+ }
+
+ // ----------------------------------------------------------------------------------------
+
+ table.free();
+ }
+
+ private void testJoin(BinaryHashTable table, HashMap<Integer, Long> map) throws IOException {
+ BinaryRow record;
+ int numBuildValues = 0;
+
+ final BaseRow probeRec = table.getCurrentProbeRow();
+ int key = probeRec.getInt(0);
+
+ RowIterator<BinaryRow> buildSide = table.getBuildSideIterator();
+ if (buildSide.advanceNext()) {
+ numBuildValues = 1;
+ record = buildSide.getRow();
+ assertEquals("Probe-side key was different than build-side key.", key, record.getInt(0));
+ } else {
+ fail("No build side values found for a probe key.");
+ }
+ while (buildSide.advanceNext()) {
+ numBuildValues++;
+ record = buildSide.getRow();
+ assertEquals("Probe-side key was different than build-side key.", key, record.getInt(0));
+ }
+
+ Long contained = map.get(key);
+ if (contained == null) {
+ contained = (long) numBuildValues;
+ } else {
+ contained = contained + numBuildValues;
+ }
+
+ map.put(key, contained);
+ }
+
+ /*
+ * This test is basically identical to the "testSpillingHashJoinWithMassiveCollisions" test, only that the number
+ * of repeated values (causing bucket collisions) are large enough to make sure that their target partition no longer
+ * fits into memory by itself and needs to be repartitioned in the recursion again.
+ */
+ @Test
+ public void testSpillingHashJoinWithTwoRecursions() throws IOException {
+ // the following two values are known to have a hash-code collision on the first recursion level.
+ // we use them to make sure one partition grows over-proportionally large
+ final int repeatedValue1 = 40559;
+ final int repeatedValue2 = 92882;
+ final int repeatedValueCountBuild = 200000;
+ final int repeatedValueCountProbe = 5;
+
+ final int numKeys = 1000000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key, plus 400k pairs with two colliding keys
+ MutableObjectIterator<BinaryRow> build1 = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+ MutableObjectIterator<BinaryRow> build2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, repeatedValueCountBuild);
+ MutableObjectIterator<BinaryRow> build3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, repeatedValueCountBuild);
+ List<MutableObjectIterator<BinaryRow>> builds = new ArrayList<>();
+ builds.add(build1);
+ builds.add(build2);
+ builds.add(build3);
+ MutableObjectIterator<BinaryRow> buildInput = new UnionIterator<>(builds);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probe1 = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+ MutableObjectIterator<BinaryRow> probe2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, 5);
+ MutableObjectIterator<BinaryRow> probe3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, 5);
+ List<MutableObjectIterator<BinaryRow>> probes = new ArrayList<>();
+ probes.add(probe1);
+ probes.add(probe2);
+ probes.add(probe3);
+ MutableObjectIterator<BinaryRow> probeInput = new UnionIterator<>(probes);
+
+ // create the map for validating the results
+ HashMap<Integer, Long> map = new HashMap<>(numKeys);
+
+ // ----------------------------------------------------------------------------------------
+ MemoryManager memManager = new MemoryManager(896 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 896 * PAGE_SIZE, ioManager);
+ final BinaryRow recordReuse = new BinaryRow(2);
+
+ BinaryRow buildRow = buildSideSerializer.createInstance();
+ while ((buildRow = buildInput.next(buildRow)) != null) {
+ table.putBuildRow(buildRow);
+ }
+ table.endBuild();
+
+ BinaryRow probeRow = probeSideSerializer.createInstance();
+ while ((probeRow = probeInput.next(probeRow)) != null) {
+ if (table.tryProbe(probeRow)){
+ testJoin(table, map);
+ }
+ }
+
+ while (table.nextMatching()){
+ testJoin(table, map);
+ }
+
+ table.close();
+
+ Assert.assertEquals("Wrong number of keys", numKeys, map.size());
+ for (Map.Entry<Integer, Long> entry : map.entrySet()) {
+ long val = entry.getValue();
+ int key = entry.getKey();
+
+ Assert.assertEquals("Wrong number of values in per-key cross product for key " + key,
+ (key == repeatedValue1 || key == repeatedValue2) ?
+ (probeValsPerKey + repeatedValueCountProbe) * (buildValsPerKey + repeatedValueCountBuild) :
+ probeValsPerKey * buildValsPerKey, val);
+ }
+
+ // ----------------------------------------------------------------------------------------
+
+ table.free();
+ }
+
+ /*
+ * This test is basically identical to the "testSpillingHashJoinWithMassiveCollisions" test, only that the number
+ * of repeated values (causing bucket collisions) are large enough to make sure that their target partition no longer
+ * fits into memory by itself and needs to be repartitioned in the recursion again.
+ */
+ @Test
+ public void testFailingHashJoinTooManyRecursions() throws IOException {
+ // the following two values are known to have a hash-code collision on the first recursion level.
+ // we use them to make sure one partition grows over-proportionally large
+ final int repeatedValue1 = 40559;
+ final int repeatedValue2 = 92882;
+ final int repeatedValueCount = 3000000;
+
+ final int numKeys = 1000000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 3 million pairs with 3 values sharing the same key, plus 400k pairs with two colliding keys
+ MutableObjectIterator<BinaryRow> build1 = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+ MutableObjectIterator<BinaryRow> build2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, repeatedValueCount);
+ MutableObjectIterator<BinaryRow> build3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, repeatedValueCount);
+ List<MutableObjectIterator<BinaryRow>> builds = new ArrayList<>();
+ builds.add(build1);
+ builds.add(build2);
+ builds.add(build3);
+ MutableObjectIterator<BinaryRow> buildInput = new UnionIterator<>(builds);
+
+ // create a probe input that gives 10 million pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probe1 = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+ MutableObjectIterator<BinaryRow> probe2 = new ConstantsKeyValuePairsIterator(repeatedValue1, 17, repeatedValueCount);
+ MutableObjectIterator<BinaryRow> probe3 = new ConstantsKeyValuePairsIterator(repeatedValue2, 23, repeatedValueCount);
+ List<MutableObjectIterator<BinaryRow>> probes = new ArrayList<>();
+ probes.add(probe1);
+ probes.add(probe2);
+ probes.add(probe3);
+ MutableObjectIterator<BinaryRow> probeInput = new UnionIterator<>(probes);
+ // ----------------------------------------------------------------------------------------
+ MemoryManager memManager = new MemoryManager(896 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 896 * PAGE_SIZE, ioManager);
+
+ try {
+ join(table, buildInput, probeInput);
+ fail("Hash Join must have failed due to too many recursions.");
+ } catch (Exception ex) {
+ // expected
+ }
+
+ table.close();
+
+ // ----------------------------------------------------------------------------------------
+
+ table.free();
+ }
+
+ /*
+ * Spills build records, so that probe records are also spilled. But only so
+ * few probe records are used that some partitions remain empty.
+ */
+ @Test
+ public void testSparseProbeSpilling() throws IOException, MemoryAllocationException {
+ final int numBuildKeys = 1000000;
+ final int numBuildVals = 1;
+ final int numProbeKeys = 20;
+ final int numProbeVals = 1;
+
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(
+ numBuildKeys, numBuildVals, false);
+ MemoryManager memManager = new MemoryManager(128 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 100 * PAGE_SIZE, ioManager);
+
+ int expectedNumResults = (Math.min(numProbeKeys, numBuildKeys) * numBuildVals)
+ * numProbeVals;
+
+ int numRecordsInJoinResult = join(table, buildInput, new UniformBinaryRowGenerator(numProbeKeys, numProbeVals, true));
+
+ Assert.assertEquals("Wrong number of records in join result.", expectedNumResults, numRecordsInJoinResult);
+
+ table.close();
+
+ table.free();
+ }
+
+ /*
+ * Same test as {@link #testSparseProbeSpilling} but using a build-side outer join
+ * that requires spilled build-side records to be returned and counted.
+ */
+ @Test
+ public void testSparseProbeSpillingWithOuterJoin() throws IOException {
+ final int numBuildKeys = 1000000;
+ final int numBuildVals = 1;
+ final int numProbeKeys = 20;
+ final int numProbeVals = 1;
+
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(
+ numBuildKeys, numBuildVals, false);
+ MemoryManager memManager = new MemoryManager(96 * PAGE_SIZE, 1);
+ final BinaryHashTable table = new BinaryHashTable(conf, new Object(),
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager, 96 * PAGE_SIZE,
+ ioManager, 24, 200000, true, HashJoinType.BUILD_OUTER, null, true, new boolean[] {true}, false);
+
+ int expectedNumResults =
+ (Math.max(numProbeKeys, numBuildKeys) * numBuildVals) * numProbeVals;
+
+ int numRecordsInJoinResult = join(table, buildInput, new UniformBinaryRowGenerator(numProbeKeys, numProbeVals, true), true);
+
+ Assert.assertEquals("Wrong number of records in join result.", expectedNumResults, numRecordsInJoinResult);
+
+ table.close();
+ table.free();
+ }
+
+ /*
+ * This test validates a bug fix against former memory loss in the case where a partition was spilled
+ * during an insert into the same.
+ */
+ @Test
+ public void validateSpillingDuringInsertion() throws IOException, MemoryAllocationException {
+ final int numBuildKeys = 500000;
+ final int numBuildVals = 1;
+ final int numProbeKeys = 10;
+ final int numProbeVals = 1;
+
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numBuildKeys, numBuildVals, false);
+ MemoryManager memManager = new MemoryManager(85 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 85 * PAGE_SIZE, ioManager);
+
+ int expectedNumResults = (Math.min(numProbeKeys, numBuildKeys) * numBuildVals)
+ * numProbeVals;
+
+ int numRecordsInJoinResult = join(table, buildInput, new UniformBinaryRowGenerator(numProbeKeys, numProbeVals, true));
+
+ Assert.assertEquals("Wrong number of records in join result.", expectedNumResults, numRecordsInJoinResult);
+
+ table.close();
+
+ table.free();
+ }
+
+ @Test
+ public void testBucketsNotFulfillSegment() throws Exception {
+ final int numKeys = 10000;
+ final int buildValsPerKey = 3;
+ final int probeValsPerKey = 10;
+
+ // create a build input that gives 30000 pairs with 3 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 100000 pairs with 10 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ // allocate the memory for the HashTable
+ MemoryManager memManager = new MemoryManager(35 * PAGE_SIZE, 1);
+ // ----------------------------------------------------------------------------------------
+
+ final BinaryHashTable table = new BinaryHashTable(conf, new Object(),
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(),
+ memManager, 35 * PAGE_SIZE, ioManager, 24, 200000,
+ true, HashJoinType.INNER, null, false, new boolean[]{true}, false);
+
+ // For FLINK-2545, the buckets data may not fulfill it's buffer, for example, the buffer may contains 256 buckets,
+ // while hash table only assign 250 bucket on it. The unused buffer bytes may contains arbitrary data, which may
+ // influence hash table if forget to skip it. To mock this, put the invalid bucket data(partition=1, inMemory=true, count=-1)
+ // at the end of buffer.
+ for (MemorySegment segment : table.getFreedMemory()) {
+ int newBucketOffset = segment.size() - 128;
+ // initialize the header fields
+ segment.put(newBucketOffset, (byte) 0);
+ segment.put(newBucketOffset + 1, (byte) 0);
+ segment.putShort(newBucketOffset + 2, (short) -1);
+ segment.putLong(newBucketOffset + 4, ~0x0L);
+ }
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput);
+
+ Assert.assertEquals("Wrong number of records in join result.", numKeys * buildValsPerKey * probeValsPerKey, numRecordsInJoinResult);
+
+ table.close();
+ table.free();
+ }
+
+ @Test
+ public void testHashWithBuildSideOuterJoin1() throws Exception {
+ final int numKeys = 20000;
+ final int buildValsPerKey = 1;
+ final int probeValsPerKey = 1;
+
+ // create a build input that gives 40000 pairs with 1 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(2 * numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 20000 pairs with 1 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+ MemoryManager memManager = new MemoryManager(35 * PAGE_SIZE, 1);
+ // allocate the memory for the HashTable
+ final BinaryHashTable table = new BinaryHashTable(conf, new Object(),
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(),
+ memManager, 35 * PAGE_SIZE, ioManager, 24, 200000, true,
+ HashJoinType.BUILD_OUTER, null, true,
+ new boolean[]{true}, false);
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput, true);
+
+ Assert.assertEquals("Wrong number of records in join result.", 2 * numKeys * buildValsPerKey * probeValsPerKey, numRecordsInJoinResult);
+
+ table.close();
+ table.free();
+ }
+
+ @Test
+ public void testHashWithBuildSideOuterJoin2() throws Exception {
+ final int numKeys = 40000;
+ final int buildValsPerKey = 2;
+ final int probeValsPerKey = 1;
+
+ // The keys of probe and build sides are overlapped, so there would be none unmatched build elements
+ // after probe phase, make sure build side outer join works well in this case.
+
+ // create a build input that gives 80000 pairs with 2 values sharing the same key
+ MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
+
+ // create a probe input that gives 40000 pairs with 1 values sharing a key
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ // allocate the memory for the HashTable
+ MemoryManager memManager = new MemoryManager(35 * PAGE_SIZE, 1);
+ final BinaryHashTable table = newBinaryHashTable(
+ this.buildSideSerializer, this.probeSideSerializer,
+ new MyProjection(), new MyProjection(), memManager,
+ 33 * PAGE_SIZE, ioManager);
+
+ // ----------------------------------------------------------------------------------------
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput, true);
+ Assert.assertEquals("Wrong number of records in join result.", numKeys * buildValsPerKey * probeValsPerKey, numRecordsInJoinResult);
+
+ table.close();
+ table.free();
+ }
+
+ @Test
+ public void testRepeatBuildJoin() throws Exception {
+ final int numKeys = 500;
+ final int probeValsPerKey = 1;
+ MemoryManager memManager = new MemoryManager(40 * PAGE_SIZE, 1);
+ MutableObjectIterator<BinaryRow> buildInput = new MutableObjectIterator<BinaryRow>() {
+
+ int cnt = 0;
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) throws IOException {
+ return next();
+ }
+
+ @Override
+ public BinaryRow next() throws IOException {
+ cnt++;
+ if (cnt > numKeys) {
+ return null;
+ }
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, 1);
+ writer.writeInt(1, 1);
+ writer.complete();
+ return row;
+ }
+ };
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ final BinaryHashTable table = new BinaryHashTable(conf, new Object(),
+ buildSideSerializer, probeSideSerializer, new MyProjection(), new MyProjection(), memManager,
+ 40 * PAGE_SIZE, ioManager, 24, 200000,
+ true, HashJoinType.INNER, null, false, new boolean[] {true}, true);
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput, true);
+ Assert.assertEquals("Wrong number of records in join result.",
+ 1, numRecordsInJoinResult);
+
+ table.close();
+ table.free();
+ }
+
+ @Test
+ public void testRepeatBuildJoinWithSpill() throws Exception {
+ final int numKeys = 30000;
+ final int numRows = 300000;
+ final int probeValsPerKey = 1;
+
+ MutableObjectIterator<BinaryRow> buildInput = new MutableObjectIterator<BinaryRow>() {
+
+ int cnt = 0;
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) throws IOException {
+ return next();
+ }
+
+ @Override
+ public BinaryRow next() throws IOException {
+ cnt++;
+ if (cnt > numRows) {
+ return null;
+ }
+ int value = cnt % numKeys;
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, value);
+ writer.writeInt(1, value);
+ writer.complete();
+ return row;
+ }
+ };
+ MemoryManager memManager = new MemoryManager(35 * PAGE_SIZE, 1);
+ MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
+
+ final BinaryHashTable table = new BinaryHashTable(
+ conf,
+ new Object(),
+ buildSideSerializer,
+ probeSideSerializer,
+ new MyProjection(),
+ new MyProjection(),
+ memManager,
+ 35 * PAGE_SIZE,
+ ioManager,
+ 24,
+ 200000,
+ true,
+ HashJoinType.INNER,
+ null,
+ false,
+ new boolean[] {true}, true);
+
+ int numRecordsInJoinResult = join(table, buildInput, probeInput, true);
+ Assert.assertTrue("Wrong number of records in join result.",
+ numRecordsInJoinResult < numRows);
+
+ table.close();
+ table.free();
+ }
+
+ // ============================================================================================
+
+ /**
+ * An iterator that returns the Key/Value pairs with identical value a given number of times.
+ */
+ public static final class ConstantsKeyValuePairsIterator implements MutableObjectIterator<BinaryRow> {
+
+ private final IntValue key;
+ private final IntValue value;
+
+ private int numLeft;
+
+ public ConstantsKeyValuePairsIterator(int key, int value, int count) {
+ this.key = new IntValue(key);
+ this.value = new IntValue(value);
+ this.numLeft = count;
+ }
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) {
+ if (this.numLeft > 0) {
+ this.numLeft--;
+
+ BinaryRowWriter writer = new BinaryRowWriter(reuse);
+ writer.writeInt(0, this.key.getValue());
+ writer.writeInt(1, this.value.getValue());
+ writer.complete();
+ return reuse;
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public BinaryRow next() {
+ return next(new BinaryRow(2));
+ }
+ }
+
+ private BinaryHashTable newBinaryHashTable(
+ BinaryRowSerializer buildSideSerializer,
+ BinaryRowSerializer probeSideSerializer,
+ Projection<BaseRow, BinaryRow> buildSideProjection,
+ Projection<BaseRow, BinaryRow> probeSideProjection,
+ MemoryManager memoryManager,
+ long memory,
+ IOManager ioManager) {
+ return new BinaryHashTable(
+ conf,
+ new Object(),
+ buildSideSerializer,
+ probeSideSerializer,
+ buildSideProjection,
+ probeSideProjection,
+ memoryManager,
+ memory,
+ ioManager,
+ 24,
+ 200000,
+ true,
+ HashJoinType.INNER,
+ null,
+ false,
+ new boolean[]{true}, false);
+ }
+
+ private static final class MyProjection implements Projection<BaseRow, BinaryRow> {
+
+ BinaryRow innerRow = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(innerRow);
+
+ @Override
+ public BinaryRow apply(BaseRow row) {
+ writer.reset();
+ if (row.isNullAt(0)) {
+ writer.setNullAt(0);
+ } else {
+ writer.writeInt(0, row.getInt(0));
+ }
+ writer.complete();
+ return innerRow;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java
index c84db7c..70beba2 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/TestMemorySegmentPool.java
@@ -45,10 +45,6 @@ public class TestMemorySegmentPool implements MemorySegmentPool {
}
@Override
- public void clear() {
- }
-
- @Override
public MemorySegment nextSegment() {
return MemorySegmentFactory.wrap(new byte[pageSize]);
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/util/UniformBinaryRowGenerator.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/util/UniformBinaryRowGenerator.java
new file mode 100644
index 0000000..e9c6d6c
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/util/UniformBinaryRowGenerator.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.util;
+
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.BinaryRowWriter;
+import org.apache.flink.types.IntValue;
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * Uniform genarator for binary row.
+ */
+public class UniformBinaryRowGenerator implements MutableObjectIterator<BinaryRow> {
+
+ int numKeys;
+ int numVals;
+ int keyCnt = 0;
+ int valCnt = 0;
+ int startKey = 0;
+ int startVal = 0;
+ boolean repeatKey;
+ private IntValue key = new IntValue();
+ private IntValue value = new IntValue();
+
+ public UniformBinaryRowGenerator(int numKeys, int numVals, boolean repeatKey) {
+ this(numKeys, numVals, 0, 0, repeatKey);
+ }
+
+ public UniformBinaryRowGenerator(
+ int numKeys, int numVals, int startKey, int startVal, boolean repeatKey) {
+ this.numKeys = numKeys;
+ this.numVals = numVals;
+ this.startKey = startKey;
+ this.startVal = startVal;
+ this.repeatKey = repeatKey;
+ }
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) {
+ if (!repeatKey) {
+ if (valCnt >= numVals + startVal) {
+ return null;
+ }
+
+ key.setValue(keyCnt++);
+ value.setValue(valCnt);
+
+ if (keyCnt == numKeys + startKey) {
+ keyCnt = startKey;
+ valCnt++;
+ }
+ } else {
+ if (keyCnt >= numKeys + startKey) {
+ return null;
+ }
+ key.setValue(keyCnt);
+ value.setValue(valCnt++);
+
+ if (valCnt == numVals + startVal) {
+ valCnt = startVal;
+ keyCnt++;
+ }
+ }
+
+ BinaryRowWriter writer = new BinaryRowWriter(reuse);
+ writer.writeInt(0, this.key.getValue());
+ writer.writeInt(1, this.value.getValue());
+ writer.complete();
+ return reuse;
+ }
+
+ @Override
+ public BinaryRow next() {
+ key = new IntValue();
+ value = new IntValue();
+ return next(new BinaryRow(2));
+ }
+}