twalthr commented on code in PR #29760:
URL: https://github.com/apache/flink/pull/29760#discussion_r4465354631
##########
docs/content/docs/connectors/table/formats/raw.md:
##########
@@ -112,12 +112,26 @@ Format Options
<td>String</td>
<td>Specify the line delimiter for splitting incoming messages into
multiple rows during
deserialization. When set, each incoming message is decoded using
'raw.charset' and then split
- by this delimiter; one row is emitted per segment. During serialization,
the delimiter bytes
+ by this delimiter; one row is emitted per segment. A
<code>VARIANT</code> column is instead split
+ on the UTF-8 bytes of the delimiter, without decoding the message.
During serialization, the delimiter bytes
Review Comment:
```suggestion
by this delimiter; one row is emitted per segment. Applies only to
STRING columns. During serialization, the delimiter bytes
```
##########
docs/content/docs/connectors/table/formats/raw.md:
##########
@@ -112,12 +112,26 @@ Format Options
<td>String</td>
<td>Specify the line delimiter for splitting incoming messages into
multiple rows during
deserialization. When set, each incoming message is decoded using
'raw.charset' and then split
- by this delimiter; one row is emitted per segment. During serialization,
the delimiter bytes
+ by this delimiter; one row is emitted per segment. A
<code>VARIANT</code> column is instead split
+ on the UTF-8 bytes of the delimiter, without decoding the message.
During serialization, the delimiter bytes
are appended to each serialized value. Common values are '\n' (newline)
or '||'.
<br><strong>Note:</strong> When the same delimiter is configured for
both serialization and
deserialization, the two are round-trip compatible: a trailing delimiter
appended by the
serializer is automatically stripped during deserialization.</td>
</tr>
+ <tr>
+ <td><h5>raw.variant-encoding</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">json</td>
+ <td>Enum</td>
+ <td>Specify how a <code>VARIANT</code> column is encoded. Valid values
are 'json' and 'binary'.
+ 'json' reads and writes a UTF-8 JSON document. It is convenient, because
people and other systems
+ can read and write JSON, but it is lossy: JSON has no type for dates,
times, timestamps, bytes or
Review Comment:
```suggestion
'json' reads and writes a UTF-8 JSON document. For convenience, and
also because JSON is a very popular and human readable format, the stored
VARIANT value is assumed to be JSON. However, JSON values are lossy: JSON has
no type for dates, times, timestamps, bytes or
```
##########
docs/content/docs/connectors/table/formats/raw.md:
##########
@@ -182,15 +196,25 @@ The table below details the SQL types the format
supports, including details of
</tr>
<tr>
<td><code>VARIANT</code></td>
- <td>A UTF-8 (by default) encoded JSON document.<br>
- The encoding charset can be configured by 'raw.charset'.<br>
- On read, the decoded text is parsed like <code>PARSE_JSON</code>, so
duplicate object keys are rejected and
- malformed JSON fails the job. On write, the value is rendered by
<code>Variant#toJson</code>. The round trip is
- value-lossless but not byte-lossless: insignificant whitespace is
dropped and object keys are ordered.</td>
+ <td>Depends on 'raw.variant-encoding'.<br>
+ <strong>json</strong> (default): a UTF-8 encoded JSON document.
+ 'raw.charset' must be UTF-8. To read or write JSON in another charset,
declare the column as
+ <code>STRING</code> and convert it with <code>PARSE_JSON</code> or
<code>JSON_STRING</code>.
+ On read, the bytes are parsed like <code>PARSE_JSON</code> with the
same checks, so an invalid UTF-8 byte such as
+ <code>0xFF</code>, a duplicate object key or malformed JSON fails the
job. On write, the value is rendered by
+ <code>Variant#toJson</code>.
Review Comment:
```suggestion
<code>VARIANT_STRING</code>.
```
are what was the SQL method to translate to JSON?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -284,29 +369,47 @@ public Object convert(byte[] data) {
}
/**
- * Creates a converter that decodes the bytes with the configured charset
and parses the text as
- * a JSON document into a {@link Variant}. Duplicate object keys are
rejected, matching the
- * default of {@code PARSE_JSON}.
+ * Creates a converter that parses UTF-8 JSON bytes into a {@link Variant}
like {@code
+ * PARSE_JSON}, so invalid UTF-8 and duplicate object keys are rejected.
*/
- private static DeserializationRuntimeConverter createVariantConverter(
- final String charsetName) {
+ private static DeserializationRuntimeConverter createVariantConverter() {
return new DeserializationRuntimeConverter() {
private static final long serialVersionUID = 1L;
- private transient Charset charset;
@Override
- public void open() {
- charset = Charset.forName(charsetName);
+ public Object convert(byte[] data) {
+ try {
+ return BinaryVariantInternalBuilder.parseJson(data, false);
+ } catch (Exception e) {
+ throw new DeserializationException(
+ "Failed to deserialize VARIANT type. "
+ + "The received data is not a valid JSON
document.",
+ e);
+ }
}
+ };
+ }
+
+ /**
+ * Creates a converter that reads the Variant binary encoding, the
metadata followed by the
+ * value. The bytes are only checked as far as needed to split them. A
malformed value fails
+ * when it is read.
+ */
+ private static DeserializationRuntimeConverter
createBinaryVariantConverter() {
+ return new DeserializationRuntimeConverter() {
+ private static final long serialVersionUID = 1L;
@Override
public Object convert(byte[] data) {
try {
- return BinaryVariantInternalBuilder.parseJson(new
String(data, charset), false);
+ final int metadataSize =
BinaryVariantUtil.metadataSize(data);
+ return new BinaryVariant(
+ Arrays.copyOfRange(data, metadataSize,
data.length),
+ Arrays.copyOf(data, metadataSize));
} catch (Exception e) {
throw new DeserializationException(
"Failed to deserialize VARIANT type. "
- + "The received data is not a valid JSON
document.",
+ + "The received data is not a valid
Variant binary encoding.",
Review Comment:
```suggestion
+ "The received data is not a valid
VARIANT binary encoding.",
```
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -87,15 +100,42 @@ public RawFormatDeserializationSchema(
String charsetName,
boolean isBigEndian,
@Nullable String lineDelimiter) {
+ this(
Review Comment:
remove overload and only keep one constructor?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -213,7 +296,9 @@ private static DeserializationRuntimeConverter
createConverter(
return RawValueData::fromBytes;
case VARIANT:
- return createVariantConverter(charsetName);
+ return variantEncoding == VariantEncoding.BINARY
+ ? createBinaryVariantConverter()
+ : createVariantConverter();
Review Comment:
```suggestion
: createJsonVariantConverter();
```
##########
docs/content/docs/connectors/table/formats/raw.md:
##########
@@ -112,12 +112,26 @@ Format Options
<td>String</td>
<td>Specify the line delimiter for splitting incoming messages into
multiple rows during
deserialization. When set, each incoming message is decoded using
'raw.charset' and then split
- by this delimiter; one row is emitted per segment. During serialization,
the delimiter bytes
+ by this delimiter; one row is emitted per segment. A
<code>VARIANT</code> column is instead split
+ on the UTF-8 bytes of the delimiter, without decoding the message.
During serialization, the delimiter bytes
Review Comment:
maybe just the positive case?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java:
##########
@@ -163,6 +177,46 @@ private static void checkFieldType(LogicalType fieldType) {
}
}
+ /**
+ * Checks the options of a VARIANT column. The binary encoding can hold
any byte, so it cannot
+ * be split by a line delimiter, and it is not text, so the charset does
not apply to it.
+ */
+ private static void checkVariantOptions(
+ LogicalType fieldType,
+ String charsetName,
+ VariantEncoding variantEncoding,
+ Optional<String> lineDelimiter) {
+ if (variantEncoding == VariantEncoding.JSON) {
+ checkVariantCharset(fieldType, charsetName);
+ } else if (fieldType.is(LogicalTypeRoot.VARIANT) &&
lineDelimiter.isPresent()) {
+ throw new ValidationException(
+ String.format(
+ "The 'raw' format does not support '%s.%s' for a
VARIANT column "
+ + "with '%s.%s' set to '%s'.",
+ IDENTIFIER,
+ RawFormatOptions.LINE_DELIMITER.key(),
+ IDENTIFIER,
+ RawFormatOptions.VARIANT_ENCODING.key(),
+ VariantEncoding.BINARY));
+ }
+ }
+
+ /**
+ * Checks that a VARIANT column uses UTF-8, which RFC 8259 requires for
JSON exchanged between
+ * systems and which {@code PARSE_JSON} reads.
+ */
+ private static void checkVariantCharset(LogicalType fieldType, String
charsetName) {
+ if (fieldType.is(LogicalTypeRoot.VARIANT)
Review Comment:
Check `fieldType.is(LogicalTypeRoot.VARIANT)` once at the beginning of
checkVariantOptions and perform an early out otherwise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
