You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "suhwan-cheon (via GitHub)" <gi...@apache.org> on 2025/09/14 05:25:13 UTC
[PR] [FLINK-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
suhwan-cheon opened a new pull request, #4117:
URL: https://github.com/apache/flink-cdc/pull/4117
issue: https://issues.apache.org/jira/browse/FLINK-38247
### Issue
An infinite loop occurred when using the **MySqlChunkSplitter** to split a table with a **MySQL BIGINT UNSIGNED primary key**. (This problem happens when the primary key value exceeds Long.MAX_VALUE)
---
### Solution
The new **`StatementUtils.setSafeObject`** method was introduced to fix this issue. It detects the overflow and correctly converts the value to a **`BigDecimal`** before setting it in the `PreparedStatement`.
---
### Verification
* A new unit test, **`StatementUtilsTest`**, was added to verify the correctness of `StatementUtils.setSafeObject`.
* Java's **dynamic proxy** (`createPreparedStatementProxy`) was used to avoid creating a verbose mock class, simplifying the test code.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "suhwan-cheon (via GitHub)" <gi...@apache.org>.
suhwan-cheon commented on PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#issuecomment-3314707258
there's a Checkstyle error. I'm not used to this environment, so I made a mistake.
I've fixed Checkstyle error and ran the MySQL connector unit tests with `mvn -pl flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc -DskipITs=true test`
please review.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2347941892
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +77,18 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint unsigned, which poses a data overflow issue.
+ // Therefore, we need to handle the overflow issue by converting the long value to BigDecimal.
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof Long && (Long) value < 0L) {
Review Comment:
For example:
```
CREATE TABLE `unsigned_bigint_pk`
--
(
`order_id` BIGINT UNSIGNED NOT NULL,
`desc` varchar(512) NOT NULL,
PRIMARY KEY (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
INSERT INTO unsigned_bigint_pk
VALUES (1, 'flink'),
(2, 'flink'),
(3, 'flink'),
(4, 'flink'),
(5, 'flink'),
(6, 'flink'),
(7, 'flink'),
(8, 'flink'),
(9, 'flink'),
(10, 'flink'),
(11, 'flink'),
(12, 'flink'),
(18446744073709551604, 'flink'),
(18446744073709551605, 'flink'),
(18446744073709551606, 'flink'),
(18446744073709551607, 'flink'),
(18446744073709551608, 'flink'),
(18446744073709551609, 'flink'),
(18446744073709551610, 'flink'),
(18446744073709551611, 'flink'),
(18446744073709551612, 'flink'),
(18446744073709551613, 'flink'),
(18446744073709551614, 'flink'),
(18446744073709551615, 'flink');
```
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2731734984
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +78,19 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint
+ // unsigned, which poses a data overflow issue.
+ // Therefore, we need to handle the overflow issue by converting the bigint unsigned to
+ // BigDecimal.
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof BigInteger) {
+ ps.setBigDecimal(parameterIndex, new BigDecimal((BigInteger) value));
+ } else {
+ ps.setObject(parameterIndex, value);
Review Comment:
The PR description says `setSafeObject` “detects the overflow”, but the implementation converts *all* `BigInteger` values to `BigDecimal` (even when the value would fit in a signed `long`). If the JDBC driver returns `BigInteger` for all BIGINT UNSIGNED values (including small ones), this changes parameter typing from BIGINT to DECIMAL for every bound, which can affect query planning/index usage. Consider only converting when `value` is outside `Long.MIN_VALUE..Long.MAX_VALUE` (or when explicitly handling unsigned overflow), and otherwise keep `setObject`/`setLong` for the common case.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "suhwan-cheon (via GitHub)" <gi...@apache.org>.
suhwan-cheon commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2359548090
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +77,18 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint unsigned, which poses a data overflow issue.
+ // Therefore, we need to handle the overflow issue by converting the long value to BigDecimal.
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof Long && (Long) value < 0L) {
Review Comment:
@lvyanquan
Hello! I added itcase in MySqlSourceITCase.
I think JDBC may surface BIGINT UNSIGNED as Long (values > Long.MAX_VALUE appear negative due to two’s complement)
So In this code, 1) detect negative Longs 2) bind them as BigInteger, ensuring values near 2^64−1 are handled correctly.
I Added an IT in MySqlSourceITCase that creates unsigned_bigint_pk and verifies boundary values. And I saw that the test worked well.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "suhwan-cheon (via GitHub)" <gi...@apache.org>.
suhwan-cheon commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2702247280
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -1447,6 +1448,98 @@ private boolean hasNextData(final CloseableIterator<?> iterator)
}
}
+ @Test
+ void testUnsignedBigintPrimaryKeyChunking() throws Exception {
+ customDatabase.createAndInitialize();
+
+ String db = customDatabase.getDatabaseName();
+ String table = "unsigned_bigint_pk";
+ try (MySqlConnection connection = getConnection()) {
+ connection.setAutoCommit(false);
+ String createSql =
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " `order_id` BIGINT UNSIGNED NOT NULL,\n"
+ + " `desc` VARCHAR(512) NOT NULL,\n"
+ + " PRIMARY KEY (`order_id`)\n"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8;",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Insert sample data including values near UNSIGNED BIGINT max
+ String insertSql =
+ String.format(
+ "INSERT INTO %s.%s (`order_id`, `desc`) VALUES "
+ + "(1, 'flink'),(2, 'flink'),(3, 'flink'),(4, 'flink'),(5, 'flink'),"
+ + "(6, 'flink'),(7, 'flink'),(8, 'flink'),(9, 'flink'),(10, 'flink'),"
+ + "(11, 'flink'),(12, 'flink'),"
+ + "(18446744073709551604, 'flink'),(18446744073709551605, 'flink'),"
+ + "(18446744073709551606, 'flink'),(18446744073709551607, 'flink'),"
+ + "(18446744073709551608, 'flink'),(18446744073709551609, 'flink'),"
+ + "(18446744073709551610, 'flink'),(18446744073709551611, 'flink'),"
+ + "(18446744073709551612, 'flink'),(18446744073709551613, 'flink'),"
+ + "(18446744073709551614, 'flink'),(18446744073709551615, 'flink');",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Drop if exists to be idempotent across runs, then create and insert
+ connection.execute(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s;",
+ StatementUtils.quote(db), StatementUtils.quote(table)),
+ createSql,
+ insertSql);
+ connection.commit();
+ }
+
+ // Build a source reading only the unsigned_bigint_pk table
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("order_id", DataTypes.DECIMAL(20, 0)),
+ DataTypes.FIELD("desc", DataTypes.STRING()));
+ LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
+ InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
+ RowDataDebeziumDeserializeSchema deserializer =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType((RowType) dataType.getLogicalType())
+ .setResultTypeInfo(typeInfo)
+ .build();
+
+ MySqlSource<RowData> source =
+ MySqlSource.<RowData>builder()
+ .hostname(MYSQL_CONTAINER.getHost())
+ .port(MYSQL_CONTAINER.getDatabasePort())
+ .username(customDatabase.getUsername())
+ .password(customDatabase.getPassword())
+ .serverTimeZone("UTC")
+ .databaseList(db)
+ .tableList(db + "." + table)
+ .deserializer(deserializer)
+ .startupOptions(StartupOptions.initial())
+ .chunkKeyColumn(new ObjectPath(db, table), "order_id")
Review Comment:
@lvyanquan
I added `.splitSize(2)` and I found my logic was wrong.
(I realized that even if I delivered BigInteger to setObject(), it would be converted to long...)
So, I modified it to use setBigDecimal() when it was BigInteger.
Spotless, test has been performed, please check 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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2366654886
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -1447,6 +1448,98 @@ private boolean hasNextData(final CloseableIterator<?> iterator)
}
}
+ @Test
+ void testUnsignedBigintPrimaryKeyChunking() throws Exception {
+ customDatabase.createAndInitialize();
+
+ String db = customDatabase.getDatabaseName();
+ String table = "unsigned_bigint_pk";
+ try (MySqlConnection connection = getConnection()) {
+ connection.setAutoCommit(false);
+ String createSql =
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " `order_id` BIGINT UNSIGNED NOT NULL,\n"
+ + " `desc` VARCHAR(512) NOT NULL,\n"
+ + " PRIMARY KEY (`order_id`)\n"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8;",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Insert sample data including values near UNSIGNED BIGINT max
+ String insertSql =
+ String.format(
+ "INSERT INTO %s.%s (`order_id`, `desc`) VALUES "
+ + "(1, 'flink'),(2, 'flink'),(3, 'flink'),(4, 'flink'),(5, 'flink'),"
+ + "(6, 'flink'),(7, 'flink'),(8, 'flink'),(9, 'flink'),(10, 'flink'),"
+ + "(11, 'flink'),(12, 'flink'),"
+ + "(18446744073709551604, 'flink'),(18446744073709551605, 'flink'),"
+ + "(18446744073709551606, 'flink'),(18446744073709551607, 'flink'),"
+ + "(18446744073709551608, 'flink'),(18446744073709551609, 'flink'),"
+ + "(18446744073709551610, 'flink'),(18446744073709551611, 'flink'),"
+ + "(18446744073709551612, 'flink'),(18446744073709551613, 'flink'),"
+ + "(18446744073709551614, 'flink'),(18446744073709551615, 'flink');",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Drop if exists to be idempotent across runs, then create and insert
+ connection.execute(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s;",
+ StatementUtils.quote(db), StatementUtils.quote(table)),
+ createSql,
+ insertSql);
+ connection.commit();
+ }
+
+ // Build a source reading only the unsigned_bigint_pk table
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("order_id", DataTypes.DECIMAL(20, 0)),
+ DataTypes.FIELD("desc", DataTypes.STRING()));
+ LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
+ InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
+ RowDataDebeziumDeserializeSchema deserializer =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType((RowType) dataType.getLogicalType())
+ .setResultTypeInfo(typeInfo)
+ .build();
+
+ MySqlSource<RowData> source =
+ MySqlSource.<RowData>builder()
+ .hostname(MYSQL_CONTAINER.getHost())
+ .port(MYSQL_CONTAINER.getDatabasePort())
+ .username(customDatabase.getUsername())
+ .password(customDatabase.getPassword())
+ .serverTimeZone("UTC")
+ .databaseList(db)
+ .tableList(db + "." + table)
+ .deserializer(deserializer)
+ .startupOptions(StartupOptions.initial())
+ .chunkKeyColumn(new ObjectPath(db, table), "order_id")
Review Comment:
You add `.splitSize(2)` here and you can set `rootLogger.level` in `log4j2-test.properties` to `INFO` to see the actual split information.
like:
```
gners.MySqlChunkSplitter - ChunkSplitter has split 2820 chunks for table customer_kgqlle.unsigned_bigint_pk
72864 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 2
72869 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 3
72871 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 4
72875 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 5
72878 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 6
72879 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 7
72881 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 8
72882 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 9
72884 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 10
72885 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 11
72990 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - ChunkSplitter has split 2830 chunks for table customer_kgqlle.unsigned_bigint_pk
72990 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 12
72992 [snapshot-splitting] INFO org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitter - Use unevenly-sized chunks for table customer_kgqlle.unsigned_bigint_pk, the chunk size is 2 from 18446744073709551604
```
Which means that this problem was not resolved.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2347941892
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +77,18 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint unsigned, which poses a data overflow issue.
+ // Therefore, we need to handle the overflow issue by converting the long value to BigDecimal.
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof Long && (Long) value < 0L) {
Review Comment:
Example of sql lists for possible test case:
```
CREATE TABLE `unsigned_bigint_pk`
--
(
`order_id` BIGINT UNSIGNED NOT NULL,
`desc` varchar(512) NOT NULL,
PRIMARY KEY (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
INSERT INTO unsigned_bigint_pk
VALUES (1, 'flink'),
(2, 'flink'),
(3, 'flink'),
(4, 'flink'),
(5, 'flink'),
(6, 'flink'),
(7, 'flink'),
(8, 'flink'),
(9, 'flink'),
(10, 'flink'),
(11, 'flink'),
(12, 'flink'),
(18446744073709551604, 'flink'),
(18446744073709551605, 'flink'),
(18446744073709551606, 'flink'),
(18446744073709551607, 'flink'),
(18446744073709551608, 'flink'),
(18446744073709551609, 'flink'),
(18446744073709551610, 'flink'),
(18446744073709551611, 'flink'),
(18446744073709551612, 'flink'),
(18446744073709551613, 'flink'),
(18446744073709551614, 'flink'),
(18446744073709551615, 'flink');
```
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#issuecomment-3290200229
Please add an itcase in MySqlSourceITCase.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "suhwan-cheon (via GitHub)" <gi...@apache.org>.
suhwan-cheon commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2551970182
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -1447,6 +1448,98 @@ private boolean hasNextData(final CloseableIterator<?> iterator)
}
}
+ @Test
+ void testUnsignedBigintPrimaryKeyChunking() throws Exception {
+ customDatabase.createAndInitialize();
+
+ String db = customDatabase.getDatabaseName();
+ String table = "unsigned_bigint_pk";
+ try (MySqlConnection connection = getConnection()) {
+ connection.setAutoCommit(false);
+ String createSql =
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " `order_id` BIGINT UNSIGNED NOT NULL,\n"
+ + " `desc` VARCHAR(512) NOT NULL,\n"
+ + " PRIMARY KEY (`order_id`)\n"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8;",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Insert sample data including values near UNSIGNED BIGINT max
+ String insertSql =
+ String.format(
+ "INSERT INTO %s.%s (`order_id`, `desc`) VALUES "
+ + "(1, 'flink'),(2, 'flink'),(3, 'flink'),(4, 'flink'),(5, 'flink'),"
+ + "(6, 'flink'),(7, 'flink'),(8, 'flink'),(9, 'flink'),(10, 'flink'),"
+ + "(11, 'flink'),(12, 'flink'),"
+ + "(18446744073709551604, 'flink'),(18446744073709551605, 'flink'),"
+ + "(18446744073709551606, 'flink'),(18446744073709551607, 'flink'),"
+ + "(18446744073709551608, 'flink'),(18446744073709551609, 'flink'),"
+ + "(18446744073709551610, 'flink'),(18446744073709551611, 'flink'),"
+ + "(18446744073709551612, 'flink'),(18446744073709551613, 'flink'),"
+ + "(18446744073709551614, 'flink'),(18446744073709551615, 'flink');",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Drop if exists to be idempotent across runs, then create and insert
+ connection.execute(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s;",
+ StatementUtils.quote(db), StatementUtils.quote(table)),
+ createSql,
+ insertSql);
+ connection.commit();
+ }
+
+ // Build a source reading only the unsigned_bigint_pk table
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("order_id", DataTypes.DECIMAL(20, 0)),
+ DataTypes.FIELD("desc", DataTypes.STRING()));
+ LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
+ InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
+ RowDataDebeziumDeserializeSchema deserializer =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType((RowType) dataType.getLogicalType())
+ .setResultTypeInfo(typeInfo)
+ .build();
+
+ MySqlSource<RowData> source =
+ MySqlSource.<RowData>builder()
+ .hostname(MYSQL_CONTAINER.getHost())
+ .port(MYSQL_CONTAINER.getDatabasePort())
+ .username(customDatabase.getUsername())
+ .password(customDatabase.getPassword())
+ .serverTimeZone("UTC")
+ .databaseList(db)
+ .tableList(db + "." + table)
+ .deserializer(deserializer)
+ .startupOptions(StartupOptions.initial())
+ .chunkKeyColumn(new ObjectPath(db, table), "order_id")
Review Comment:
Sorry for the late reply.
I've checked that there's a problem, and I'll try to fix it as soon as possible!
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2850604900
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtilsTest.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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://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.cdc.connectors.mysql.source.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit test for {@link org.apache.flink.cdc.connectors.mysql.source.utils.StatementUtils}. */
+class StatementUtilsTest {
+
+ @Test
+ void testSetSafeObjectConvertsBigIntegerToBigDecimal() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Create a BigInteger value that exceeds Long.MAX_VALUE
+ BigInteger bigIntValue = new BigInteger("9223372036854775808"); // Long.MAX_VALUE + 1
+ BigDecimal expectedBigDecimal = new BigDecimal(bigIntValue);
+
+ // Use the safe method
+ StatementUtils.setSafeObject(psProxy, 1, bigIntValue);
+
+ // Assert that it correctly used setBigDecimal with the converted BigDecimal value
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setBigDecimal");
+ assertThat(invocationDetails.get("value")).isInstanceOf(BigDecimal.class);
+ assertThat(invocationDetails.get("value")).isEqualTo(expectedBigDecimal);
+ }
+
+ @Test
+ void testSetSafeObjectHandlesLargeBigIntegerValues() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Test with BIGINT UNSIGNED max value
+ BigInteger maxUnsignedBigInt = new BigInteger("18446744073709551615"); // 2^64 - 1
+ BigDecimal expectedBigDecimal = new BigDecimal(maxUnsignedBigInt);
+
+ StatementUtils.setSafeObject(psProxy, 1, maxUnsignedBigInt);
+
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setBigDecimal");
+ assertThat(invocationDetails.get("value")).isEqualTo(expectedBigDecimal);
+ }
+
+ @Test
+ void testSetSafeObjectHandlesRegularValues() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Test with a common Long
+ StatementUtils.setSafeObject(psProxy, 1, 123L);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo(123L);
+ invocationDetails.clear();
+
+ // Test with a String
+ StatementUtils.setSafeObject(psProxy, 2, "test");
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo("test");
+ invocationDetails.clear();
+
+ // Test with an Integer
+ StatementUtils.setSafeObject(psProxy, 3, 456);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo(456);
+ invocationDetails.clear();
+
+ // Test with null
+ StatementUtils.setSafeObject(psProxy, 4, null);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isNull();
+ }
Review Comment:
The test doesn't verify the behavior for BigInteger values that are within the Long range (between Long.MIN_VALUE and Long.MAX_VALUE inclusive). According to the implementation in setSafeObject, these values are converted to long before being set via setObject. Add a test case to verify this behavior, for example testing with BigInteger.valueOf(Long.MAX_VALUE) to ensure it correctly uses setObject with a long value.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +78,24 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint
+ // unsigned, which poses a data overflow issue for values exceeding Long.MAX_VALUE.
+ // Therefore, we need to convert to BigDecimal when the value is outside the long range
Review Comment:
The comment should follow Javadoc format for consistency with other public methods in this class. Consider converting this to a proper Javadoc comment with `@param` and `@throws` tags.
```suggestion
/**
* Safely sets an object parameter on a {@link PreparedStatement}, handling unsigned BIGINT
* values to avoid overflow. {@code PreparedStatement#setObject} converts {@link BigInteger}
* values to {@code long}, which can overflow for values outside the {@code long} range; this
* method converts such values to {@link BigDecimal} instead.
*
* @param ps the prepared statement on which the parameter is to be set
* @param parameterIndex the index of the parameter to set
* @param value the parameter value; {@link BigInteger} values outside the {@code long} range
* are converted to {@link BigDecimal}
* @throws SQLException if setting the parameter on the prepared statement fails
*/
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtilsTest.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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://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.cdc.connectors.mysql.source.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit test for {@link org.apache.flink.cdc.connectors.mysql.source.utils.StatementUtils}. */
+class StatementUtilsTest {
+
+ @Test
+ void testSetSafeObjectConvertsBigIntegerToBigDecimal() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Create a BigInteger value that exceeds Long.MAX_VALUE
+ BigInteger bigIntValue = new BigInteger("9223372036854775808"); // Long.MAX_VALUE + 1
+ BigDecimal expectedBigDecimal = new BigDecimal(bigIntValue);
+
+ // Use the safe method
+ StatementUtils.setSafeObject(psProxy, 1, bigIntValue);
+
+ // Assert that it correctly used setBigDecimal with the converted BigDecimal value
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setBigDecimal");
+ assertThat(invocationDetails.get("value")).isInstanceOf(BigDecimal.class);
+ assertThat(invocationDetails.get("value")).isEqualTo(expectedBigDecimal);
+ }
+
+ @Test
+ void testSetSafeObjectHandlesLargeBigIntegerValues() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Test with BIGINT UNSIGNED max value
+ BigInteger maxUnsignedBigInt = new BigInteger("18446744073709551615"); // 2^64 - 1
+ BigDecimal expectedBigDecimal = new BigDecimal(maxUnsignedBigInt);
+
+ StatementUtils.setSafeObject(psProxy, 1, maxUnsignedBigInt);
+
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setBigDecimal");
+ assertThat(invocationDetails.get("value")).isEqualTo(expectedBigDecimal);
+ }
+
+ @Test
+ void testSetSafeObjectHandlesRegularValues() throws SQLException {
+ Map<String, Object> invocationDetails = new HashMap<>();
+ PreparedStatement psProxy = createPreparedStatementProxy(invocationDetails);
+
+ // Test with a common Long
+ StatementUtils.setSafeObject(psProxy, 1, 123L);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo(123L);
+ invocationDetails.clear();
+
+ // Test with a String
+ StatementUtils.setSafeObject(psProxy, 2, "test");
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo("test");
+ invocationDetails.clear();
+
+ // Test with an Integer
+ StatementUtils.setSafeObject(psProxy, 3, 456);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isEqualTo(456);
+ invocationDetails.clear();
+
+ // Test with null
+ StatementUtils.setSafeObject(psProxy, 4, null);
+ assertThat(invocationDetails.get("methodName")).isEqualTo("setObject");
+ assertThat(invocationDetails.get("value")).isNull();
+ }
Review Comment:
The test cases should include a test for negative BigInteger values to ensure they are handled correctly. While MySQL BIGINT UNSIGNED cannot have negative values, the code handles negative BigInteger values that are less than Long.MIN_VALUE. Add a test case with a value like BigInteger.valueOf(Long.MIN_VALUE).subtract(BigInteger.ONE) to verify this edge case works as expected.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +78,24 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint
+ // unsigned, which poses a data overflow issue for values exceeding Long.MAX_VALUE.
+ // Therefore, we need to convert to BigDecimal when the value is outside the long range
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof BigInteger) {
+ BigInteger bigIntValue = (BigInteger) value;
+ if (bigIntValue.compareTo(BigInteger.valueOf(Long.MAX_VALUE)) > 0
+ || bigIntValue.compareTo(BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
+ ps.setBigDecimal(parameterIndex, new BigDecimal(bigIntValue));
+ } else {
+ ps.setObject(parameterIndex, bigIntValue.longValueExact());
+ }
Review Comment:
The conversion to long using longValueExact() is safe for BigInteger values within the Long range, but this creates an inconsistency in behavior. When a BigInteger is within the Long range, it's converted to a long before being set, which means the JDBC driver will handle it as a Long. When it's outside the range, it's converted to BigDecimal. This inconsistency could lead to subtle differences in how MySQL interprets these values, especially for unsigned columns. Consider always using BigDecimal for BigInteger values to ensure consistent behavior regardless of the magnitude.
```suggestion
// unsigned, which poses a data overflow issue for values exceeding Long.MAX_VALUE and can
// also lead to inconsistent behavior between in-range and out-of-range BigInteger values.
// To ensure consistent handling, always bind BigInteger values as BigDecimal.
public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
throws SQLException {
if (value instanceof BigInteger) {
BigInteger bigIntValue = (BigInteger) value;
ps.setBigDecimal(parameterIndex, new BigDecimal(bigIntValue));
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -1447,6 +1448,99 @@ private boolean hasNextData(final CloseableIterator<?> iterator)
}
}
+ @Test
+ void testUnsignedBigintPrimaryKeyChunking() throws Exception {
+ customDatabase.createAndInitialize();
+
+ String db = customDatabase.getDatabaseName();
+ String table = "unsigned_bigint_pk";
+ try (MySqlConnection connection = getConnection()) {
+ connection.setAutoCommit(false);
+ String createSql =
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " `order_id` BIGINT UNSIGNED NOT NULL,\n"
+ + " `desc` VARCHAR(512) NOT NULL,\n"
+ + " PRIMARY KEY (`order_id`)\n"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8;",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Insert sample data including values near UNSIGNED BIGINT max
+ String insertSql =
+ String.format(
+ "INSERT INTO %s.%s (`order_id`, `desc`) VALUES "
+ + "(1, 'flink'),(2, 'flink'),(3, 'flink'),(4, 'flink'),(5, 'flink'),"
+ + "(6, 'flink'),(7, 'flink'),(8, 'flink'),(9, 'flink'),(10, 'flink'),"
+ + "(11, 'flink'),(12, 'flink'),"
+ + "(18446744073709551604, 'flink'),(18446744073709551605, 'flink'),"
+ + "(18446744073709551606, 'flink'),(18446744073709551607, 'flink'),"
+ + "(18446744073709551608, 'flink'),(18446744073709551609, 'flink'),"
+ + "(18446744073709551610, 'flink'),(18446744073709551611, 'flink'),"
+ + "(18446744073709551612, 'flink'),(18446744073709551613, 'flink'),"
+ + "(18446744073709551614, 'flink'),(18446744073709551615, 'flink');",
+ StatementUtils.quote(db), StatementUtils.quote(table));
+ // Drop if exists to be idempotent across runs, then create and insert
+ connection.execute(
+ String.format(
+ "DROP TABLE IF EXISTS %s.%s;",
+ StatementUtils.quote(db), StatementUtils.quote(table)),
+ createSql,
+ insertSql);
+ connection.commit();
+ }
+
+ // Build a source reading only the unsigned_bigint_pk table
+ DataType dataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("order_id", DataTypes.DECIMAL(20, 0)),
+ DataTypes.FIELD("desc", DataTypes.STRING()));
+ LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
+ InternalTypeInfo<RowData> typeInfo = InternalTypeInfo.of(logicalType);
+ RowDataDebeziumDeserializeSchema deserializer =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType((RowType) dataType.getLogicalType())
+ .setResultTypeInfo(typeInfo)
+ .build();
+
+ MySqlSource<RowData> source =
+ MySqlSource.<RowData>builder()
+ .hostname(MYSQL_CONTAINER.getHost())
+ .port(MYSQL_CONTAINER.getDatabasePort())
+ .username(customDatabase.getUsername())
+ .password(customDatabase.getPassword())
+ .serverTimeZone("UTC")
+ .databaseList(db)
+ .tableList(db + "." + table)
+ .deserializer(deserializer)
+ .startupOptions(StartupOptions.initial())
+ .chunkKeyColumn(new ObjectPath(db, table), "order_id")
+ .splitSize(2)
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ try (CloseableIterator<RowData> it =
+ env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
+ .executeAndCollect()) {
+ // Expect 24 records as inserted above
Review Comment:
The test comment states "Expect 24 records as inserted above" but the actual insert statement inserts 12 small values (1-12) and 12 large values (18446744073709551604-18446744073709551615), which indeed totals 24. However, the comment should be more explicit about what records are expected to help with debugging if the test fails. Consider adding a more descriptive comment or using a named constant for the expected count.
```suggestion
// Expect 24 records: 12 small PKs (1–12) and 12 large PKs
// (18446744073709551604–18446744073709551615) inserted in unsigned_bigint_pk
```
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#discussion_r2347724744
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/StatementUtils.java:
##########
@@ -76,6 +77,18 @@ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
});
}
+ // PreparedStatement#setObject method will be converted to long type when handling bigint unsigned, which poses a data overflow issue.
+ // Therefore, we need to handle the overflow issue by converting the long value to BigDecimal.
+ public static void setSafeObject(PreparedStatement ps, int parameterIndex, Object value)
+ throws SQLException {
+ if (value instanceof Long && (Long) value < 0L) {
Review Comment:
It should be BigInteger type, you can verify this in itcase.
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "suhwan-cheon (via GitHub)" <gi...@apache.org>.
suhwan-cheon commented on PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#issuecomment-3311410205
> Run 'mvn spotless:apply' to fix these violations.
I've applied it. Can you run the test again?
--
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-38247] Handle BIGINT UNSIGNED overflow in PreparedStatement [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4117:
URL: https://github.com/apache/flink-cdc/pull/4117#issuecomment-3311285075
Run 'mvn spotless:apply' to fix these violations.
--
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