You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "snuyanzin (via GitHub)" <gi...@apache.org> on 2026/09/28 10:19:06 UTC
[PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
snuyanzin opened a new pull request, #29313:
URL: https://github.com/apache/flink/pull/29313
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` and replace the placeholder in the "Generated-by"
line with the tool name and version. Otherwise remove the "Generated-by" line.
See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed. See
AGENTS.md for the full guidance.
-->
- [ ] Yes (please specify the tool below)
Generated-by: [Tool Name and Version]
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4152617722
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,14 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
yes, reuse makes sense...
however in long term probably need to merge these 2 classes, no idea why we need both 🤷
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29313:
URL: https://github.com/apache/flink/pull/29313#issuecomment-5868182324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3ff433d21be627846bb85d3a3ce8781dd1b5da27",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ff433d21be627846bb85d3a3ce8781dd1b5da27",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ff433d21be627846bb85d3a3ce8781dd1b5da27 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin closed pull request #29313: [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils
URL: https://github.com/apache/flink/pull/29313
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4164981237
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,15 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // Delegate to the optimized table-common impl so non-ASCII input doesn't regress.
+ return org.apache.flink.table.data.binary.StringUtf8Utils.decodeUTF8(
+ input, offset, byteLen);
}
public static int decodeUTF8Strict(byte[] sa, int sp, int len, char[] da) {
Review Comment:
Can we merge the two StringUtf8Utils into one class? I find it confusing to have the logic split across 2 classes and mutally depend on each other.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,15 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
Review Comment:
Drop the comment, it is not relevant anymore.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4165619175
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,15 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // Delegate to the optimized table-common impl so non-ASCII input doesn't regress.
+ return org.apache.flink.table.data.binary.StringUtf8Utils.decodeUTF8(
+ input, offset, byteLen);
}
public static int decodeUTF8Strict(byte[] sa, int sp, int len, char[] da) {
Review Comment:
dropped one class in a separate commit
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4133862283
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,117 +18,25 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
thanks for the feedback
that's true
however I found another way: I added SWAR check and depending on result it goes ascii or non ascii way.
I this case there is regression for ascii for very short strings (1-5 chars), then similar for ~ 10 and after 50 there is noticeable improvement
for non ascii timings are similar to before changes
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4152612255
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,119 +18,162 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
+ /** Reads 8 bytes at a time from a {@code byte[]} for the SWAR ASCII scan. */
+ private static final VarHandle LONG_VIEW =
+ MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.nativeOrder());
+
+ /** High bit of each byte in a 64-bit word; a set bit marks a non-ASCII byte. */
+ private static final long ASCII_MASK = 0x8080808080808080L;
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
+ return str.getBytes(StandardCharsets.UTF_8);
+ }
+
+ public static String decodeUTF8(byte[] input, int offset, int byteLen) {
+ // Most real text is ASCII: route it to the JDK's compact-string path, which is a large win
+ // for longer strings. Anything with a multibyte sequence goes through the hand-rolled
+ // decoder; it beats the CharsetDecoder path for non-ASCII input.
+ if (isAscii(input, offset, byteLen)) {
+ // Pure ASCII: bytes map 1:1 to chars; ISO-8859-1 is an intrinsic LATIN1 copy.
+ return new String(input, offset, byteLen, StandardCharsets.ISO_8859_1);
+ }
+ char[] chars = allocateReuseChars(byteLen);
+ int len = decodeUTF8Strict(input, offset, byteLen, chars);
+ if (len < 0) {
+ // Malformed input; map to U+FFFD via the JDK decoder (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
+ }
+ return new String(chars, 0, len);
+ }
+
+ /**
+ * SWAR ASCII test: reads 8 bytes per step and checks the high bit of each via {@link
+ * #ASCII_MASK}, so a fully ASCII range is scanned ~8x faster than byte-by-byte and non-ASCII
+ * input bails at the first eight-byte block that contains a set high bit.
+ */
+ private static boolean isAscii(byte[] bytes, int offset, int len) {
+ int i = offset;
+ final int end = offset + len;
+ final int swarEnd = offset + (len & ~7);
+ while (i < swarEnd) {
+ if (((long) LONG_VIEW.get(bytes, i) & ASCII_MASK) != 0) {
+ return false;
Review Comment:
added some tests here
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4129863303
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,117 +18,25 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
JMH on the codec alone: Latin-1 decodes at 0.63x of today on JDK 11 and 17, short Latin-1 and CJK strings at 0.3x to 0.6x on 21 too, ASCII 1.5x to 8x faster. I'd keep the strict decoder unless you have non-ASCII numbers that say 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4135273729
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,117 +18,25 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
Thanks, verified: same output as the old decoder on 22M inputs, and non-ASCII is back at 0.8x to 1.1x of master on 11, 17 and 21.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,14 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
This copy still uses `new String(.., UTF_8)`, and PyFlink's `StringSerializer` calls it per record, so non-ASCII still regresses there. Can it delegate to the table-common `decodeUTF8`?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,119 +18,162 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
+ /** Reads 8 bytes at a time from a {@code byte[]} for the SWAR ASCII scan. */
+ private static final VarHandle LONG_VIEW =
+ MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.nativeOrder());
+
+ /** High bit of each byte in a 64-bit word; a set bit marks a non-ASCII byte. */
+ private static final long ASCII_MASK = 0x8080808080808080L;
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
+ return str.getBytes(StandardCharsets.UTF_8);
+ }
+
+ public static String decodeUTF8(byte[] input, int offset, int byteLen) {
+ // Most real text is ASCII: route it to the JDK's compact-string path, which is a large win
+ // for longer strings. Anything with a multibyte sequence goes through the hand-rolled
+ // decoder; it beats the CharsetDecoder path for non-ASCII input.
+ if (isAscii(input, offset, byteLen)) {
+ // Pure ASCII: bytes map 1:1 to chars; ISO-8859-1 is an intrinsic LATIN1 copy.
+ return new String(input, offset, byteLen, StandardCharsets.ISO_8859_1);
+ }
+ char[] chars = allocateReuseChars(byteLen);
+ int len = decodeUTF8Strict(input, offset, byteLen, chars);
+ if (len < 0) {
+ // Malformed input; map to U+FFFD via the JDK decoder (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
+ }
+ return new String(chars, 0, len);
+ }
+
+ /**
+ * SWAR ASCII test: reads 8 bytes per step and checks the high bit of each via {@link
+ * #ASCII_MASK}, so a fully ASCII range is scanned ~8x faster than byte-by-byte and non-ASCII
+ * input bails at the first eight-byte block that contains a set high bit.
+ */
+ private static boolean isAscii(byte[] bytes, int offset, int len) {
+ int i = offset;
+ final int end = offset + len;
+ final int swarEnd = offset + (len & ~7);
+ while (i < swarEnd) {
+ if (((long) LONG_VIEW.get(bytes, i) & ASCII_MASK) != 0) {
+ return false;
Review Comment:
No test reaches this `return false`: with the check disabled, `StringUtf8UtilsTest` and `BinaryStringDataTest` still pass. Decoding `Café au lait` from bytes catches it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Bytes [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4129863303
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,117 +18,25 @@
package org.apache.flink.table.data.binary;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
-import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
-import static org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
/** Utilities for String UTF-8. */
@Internal
public final class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
private StringUtf8Utils() {
// do not instantiate
}
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
JMH on the codec alone: Latin-1 decodes at 0.63x of today on JDK 11 and 17, short Latin-1 and CJK strings at 0.3x to 0.6x on 21 too, ASCII 1.5x to 8x faster.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4165620749
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,14 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
dropped one class in a separate commit
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #29313:
URL: https://github.com/apache/flink/pull/29313#issuecomment-5956530303
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40830][table] Reuse java's getBytes in StringUtf8Utils [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4153964499
##########
flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java:
##########
@@ -86,6 +87,33 @@ void testFromUtf8Bytes() {
.hasMessageContaining("Invalid UTF-8 byte at index 2 of 3");
}
+ @Test
+ void testDecodeUtf8RoundTrip() {
Review Comment:
This runs the same four inputs as the parameterized `testDecodeUtf8RoundTrip(String)` below, so one of the two can go.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,14 @@
*/
public class StringUtf8Utils {
- private static final int MAX_BYTES_PER_CHAR = 3;
-
/** This method must have the same result with JDK's String.getBytes. */
public static byte[] encodeUTF8(String str) {
- byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
- int len = encodeUTF8(str, bytes);
- return Arrays.copyOf(bytes, len);
- }
-
- public static int encodeUTF8(String str, byte[] bytes) {
- int offset = 0;
- int len = str.length();
- int sl = offset + len;
- int dp = 0;
- int dlASCII = dp + Math.min(len, bytes.length);
-
- // ASCII only optimized loop
- while (dp < dlASCII && str.charAt(offset) < '\u0080') {
- bytes[dp++] = (byte) str.charAt(offset++);
- }
-
- while (offset < sl) {
- char c = str.charAt(offset++);
- if (c < 0x80) {
- // Have at most seven bits
- bytes[dp++] = (byte) c;
- } else if (c < 0x800) {
- // 2 bytes, 11 bits
- bytes[dp++] = (byte) (0xc0 | (c >> 6));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- } else if (Character.isSurrogate(c)) {
- final int uc;
- int ip = offset - 1;
- if (Character.isHighSurrogate(c)) {
- if (sl - ip < 2) {
- uc = -1;
- } else {
- char d = str.charAt(ip + 1);
- if (Character.isLowSurrogate(d)) {
- uc = Character.toCodePoint(c, d);
- } else {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- }
- }
- } else {
- if (Character.isLowSurrogate(c)) {
- // for some illegal character
- // the jdk will ignore the origin character and cast it to '?'
- // this acts the same with jdk
- return defaultEncodeUTF8(str, bytes);
- } else {
- uc = c;
- }
- }
-
- if (uc < 0) {
- bytes[dp++] = (byte) '?';
- } else {
- bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
- bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
- bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
- offset++; // 2 chars
- }
- } else {
- // 3 bytes, 16 bits
- bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
- bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
- bytes[dp++] = (byte) (0x80 | (c & 0x3f));
- }
- }
- return dp;
- }
-
- public static int defaultEncodeUTF8(String str, byte[] bytes) {
- try {
- byte[] buffer = str.getBytes("UTF-8");
- System.arraycopy(buffer, 0, bytes, 0, buffer.length);
- return buffer.length;
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("encodeUTF8 error", e);
- }
+ return str.getBytes(StandardCharsets.UTF_8);
}
public static String decodeUTF8(byte[] input, int offset, int byteLen) {
- char[] chars = allocateReuseChars(byteLen);
- int len = decodeUTF8Strict(input, offset, byteLen, chars);
- if (len < 0) {
- return defaultDecodeUTF8(input, offset, byteLen);
- }
- return new String(chars, 0, len);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
Agreed. The runtime one was copied to table-common with FLINK-16996 and just never deleted. Either a separate commit here or a follow-up ticket works for me, what do you prefer?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org