Thank you for taking my email.
TextInputFormat.setCharsetName("UTF-16") appears to set the private variable TextInputFormat.charsetName.
It doesn't appear to cause additional behavior that would help interpret UTF-16 data.
The method I've tested is calling DelimitedInputFormat.setCharset("UTF-16"), which then sets TextInputFormat.charsetName and then modifies the previously set delimiterString to construct the proper byte string encoding of the the delimiter. This same charsetName is also used in TextInputFormat.readRecord() to interpret the bytes read from the file.
There are two problems that this implementation would seem to have when using UTF-16.
- delimiterString.getBytes(getCharset()) in DelimitedInputFormat.java will return a Big Endian byte sequence including the Byte Order Mark (BOM). The actual text file will not contain a BOM at each line ending, so the delimiter will never be read. Moreover, if the actual byte encoding of the file is Little Endian, the bytes will be interpreted incorrectly.
- TextInputFormat.readRecord() will not see a BOM each time it decodes a byte sequence with the String(bytes, offset, numBytes, charset) call. Therefore, it will assume Big Endian, which may not always be correct.
While there are likely many solutions, I would think that all of them would have to start by reading the BOM from the file when a Split is opened and then using that BOM to modify the specified encoding to a BOM specific one when the caller doesn't specify one, and to overwrite the caller's specification if the BOM is in conflict with the caller's specification. That is, if the BOM indicates Little Endian and the caller indicates UTF-16BE, Flink should rewrite the charsetName as UTF-16LE.
I hope this makes sense and that I haven't been testing incorrectly or misreading the code.