Revert "[FLINK-12879][Table SQL / Runtime] Restore the logic for writ…

…ing string" This reverts commit 9effbab.
apache · Jul 1, 2019 · a938dd7 · a938dd7
1 parent 9effbab
commit a938dd7
Showing 1 changed file with 22 additions and 3 deletions.
diff --git a/...e-runtime-blink/src/main/java/org/apache/flink/table/dataformat/AbstractBinaryWriter.java b/...e-runtime-blink/src/main/java/org/apache/flink/table/dataformat/AbstractBinaryWriter.java
@@ -78,9 +78,7 @@ public void writeString(int pos, BinaryString input) {
 		} else {
 			int len = input.getSizeInBytes();
 			if (len <= MAX_FIX_PART_DATA_SIZE) {
-				byte[] bytes = SegmentsUtil.allocateReuseBytes(len);
-				SegmentsUtil.copyToBytes(input.getSegments(), input.getOffset(), bytes, 0, len);
-				writeBytesToFixLenPart(segment, getFieldOffset(pos), bytes, len);
+				writeBytesToFixLenPart(segment, getFieldOffset(pos), input.getSegments(), input.getOffset(), len);
 			} else {
 				writeSegmentsToVarLenPart(pos, input.getSegments(), input.getOffset(), len);
 			}
@@ -302,6 +300,27 @@ private static void writeBytesToFixLenPart(
 		segment.putLong(fieldOffset, offsetAndSize);
 	}
 
+	private static void writeBytesToFixLenPart(
+		MemorySegment segment, int fieldOffset, MemorySegment[] srcSegments, int srcOffset, int len) {
+		long firstByte = len | 0x80; // first bit is 1, other bits is len
+		long sevenBytes = 0L; // real data
+		if (BinaryRow.LITTLE_ENDIAN) {
+			for (int i = 0; i < len; i++) {
+				byte b = SegmentsUtil.getByte(srcSegments, srcOffset + i);
+				sevenBytes |= ((0x00000000000000FFL & b) << (i * 8L));
+			}
+		} else {
+			for (int i = 0; i < len; i++) {
+				byte b = SegmentsUtil.getByte(srcSegments, srcOffset + i);
+				sevenBytes |= ((0x00000000000000FFL & b) << ((6 - i) * 8L));
+			}
+		}
+
+		final long offsetAndSize = (firstByte << 56) | sevenBytes;
+
+		segment.putLong(fieldOffset, offsetAndSize);
+	}
+
 	/**
 	 * OutputView for write Generic.
 	 */