[FLINK-11856][table-runtime-blink] Introduce BinaryHashTable to batch…

… table runtime. (apache#7949)

flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java

@@ -1426,4 +1426,12 @@ public final boolean equalTo(MemorySegment seg2, int offset1, int offset2, int l
 
  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;
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/memory/AbstractPagedOutputView.java

@@ -415,4 +415,35 @@ public void write(DataInputView source, int numBytes) throws IOException {
  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;
+ }
+ }
+ }
+ }
 }

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:https://www.apache.org/licenses/LICENSE-2.0
+ *  http:https://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")

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:https://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;
+ }
+
+}

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/JoinCondition.java

@@ -0,0 +1,33 @@
+/*
+ * 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:https://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.generated;
+
+import org.apache.flink.table.dataformat.BaseRow;
+
+/**
+ * Interface for code generated condition function for [[org.apache.calcite.rel.core.Join]].
+ */
+public interface JoinCondition {
+
+ /**
+ * @return true if the join condition stays true for the joined row (in1, in2)
+ */
+ boolean apply(BaseRow in1, BaseRow in2);
+
+}

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/generated/Projection.java

@@ -0,0 +1,30 @@
+/*
+ * 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:https://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.generated;
+
+import org.apache.flink.table.dataformat.BaseRow;
+
+/**
+ * Interface for code generated projection, which will map a BaseRow to another one.
+ */
+public interface Projection<IN extends BaseRow, OUT extends BaseRow> {
+
+ OUT apply(IN row);
+
+}