Skip to content

Commit 51e4ab9

Browse files
eskabetxeeskabetxe
committed
[FLINK-34883] Fix postgres uuid column as PK
1 parent d97124f commit 51e4ab9

File tree

11 files changed

+162
-118
lines changed

11 files changed

+162
-118
lines changed

‎flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -47,11 +47,10 @@ public interface JdbcSourceChunkSplitter extends ChunkSplitter {
4747
*
4848
* @param jdbc JDBC connection.
4949
* @param tableId table identity.
50-
* @param columnName column name.
50+
* @param column column name.
5151
* @return maximum and minimum value.
5252
*/
53-
Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
54-
throws SQLException;
53+
Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, Column column) throws SQLException;
5554

5655
/**
5756
* Query the minimum value of the column in the table, and the minimum value must greater than
@@ -60,12 +59,11 @@ Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
6059
*
6160
* @param jdbc JDBC connection.
6261
* @param tableId table identity.
63-
* @param columnName column name.
62+
* @param column column name.
6463
* @param excludedLowerBound the minimum value should be greater than this value.
6564
* @return minimum value.
6665
*/
67-
Object queryMin(
68-
JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
66+
Object queryMin(JdbcConnection jdbc, TableId tableId, Column column, Object excludedLowerBound)
6967
throws SQLException;
7068

7169
/**
@@ -75,15 +73,15 @@ Object queryMin(
7573
*
7674
* @param jdbc JDBC connection.
7775
* @param tableId table identity.
78-
* @param columnName column name.
76+
* @param column column name.
7977
* @param chunkSize chunk size.
8078
* @param includedLowerBound the previous chunk end value.
8179
* @return next chunk end value.
8280
*/
8381
Object queryNextChunkMax(
8482
JdbcConnection jdbc,
8583
TableId tableId,
86-
String columnName,
84+
Column column,
8785
int chunkSize,
8886
Object includedLowerBound)
8987
throws SQLException;

‎flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/experimental/MySqlChunkSplitter.java‎

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -110,28 +110,28 @@ public Collection<SnapshotSplit> generateSplits(TableId tableId) {
110110
}
111111

112112
@Override
113-
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
113+
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, Column column)
114114
throws SQLException {
115-
return MySqlUtils.queryMinMax(jdbc, tableId, columnName);
115+
return MySqlUtils.queryMinMax(jdbc, tableId, column.name());
116116
}
117117

118118
@Override
119119
public Object queryMin(
120-
JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
120+
JdbcConnection jdbc, TableId tableId, Column column, Object excludedLowerBound)
121121
throws SQLException {
122-
return MySqlUtils.queryMin(jdbc, tableId, columnName, excludedLowerBound);
122+
return MySqlUtils.queryMin(jdbc, tableId, column.name(), excludedLowerBound);
123123
}
124124

125125
@Override
126126
public Object queryNextChunkMax(
127127
JdbcConnection jdbc,
128128
TableId tableId,
129-
String columnName,
129+
Column column,
130130
int chunkSize,
131131
Object includedLowerBound)
132132
throws SQLException {
133133
return MySqlUtils.queryNextChunkMax(
134-
jdbc, tableId, columnName, chunkSize, includedLowerBound);
134+
jdbc, tableId, column.name(), chunkSize, includedLowerBound);
135135
}
136136

137137
@Override
@@ -161,8 +161,7 @@ public DataType fromDbzColumn(Column splitColumn) {
161161
*/
162162
private List<ChunkRange> splitTableIntoChunks(
163163
JdbcConnection jdbc, TableId tableId, Column splitColumn) throws SQLException {
164-
final String splitColumnName = splitColumn.name();
165-
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumnName);
164+
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumn);
166165
final Object min = minMax[0];
167166
final Object max = minMax[1];
168167
if (min == null || max == null || min.equals(max)) {
@@ -189,11 +188,10 @@ private List<ChunkRange> splitTableIntoChunks(
189188
return splitEvenlySizedChunks(
190189
tableId, min, max, approximateRowCnt, chunkSize, dynamicChunkSize);
191190
} else {
192-
return splitUnevenlySizedChunks(
193-
jdbc, tableId, splitColumnName, min, max, chunkSize);
191+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
194192
}
195193
} else {
196-
return splitUnevenlySizedChunks(jdbc, tableId, splitColumnName, min, max, chunkSize);
194+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
197195
}
198196
}
199197

@@ -241,7 +239,7 @@ private List<ChunkRange> splitEvenlySizedChunks(
241239
private List<ChunkRange> splitUnevenlySizedChunks(
242240
JdbcConnection jdbc,
243241
TableId tableId,
244-
String splitColumnName,
242+
Column splitColumn,
245243
Object min,
246244
Object max,
247245
int chunkSize)
@@ -250,15 +248,15 @@ private List<ChunkRange> splitUnevenlySizedChunks(
250248
"Use unevenly-sized chunks for table {}, the chunk size is {}", tableId, chunkSize);
251249
final List<ChunkRange> splits = new ArrayList<>();
252250
Object chunkStart = null;
253-
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumnName, max, chunkSize);
251+
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumn, max, chunkSize);
254252
int count = 0;
255253
while (chunkEnd != null && ObjectUtils.compare(chunkEnd, max) <= 0) {
256254
// we start from [null, min + chunk_size) and avoid [null, min)
257255
splits.add(ChunkRange.of(chunkStart, chunkEnd));
258256
// may sleep a while to avoid DDOS on MySQL server
259257
maySleep(count++, tableId);
260258
chunkStart = chunkEnd;
261-
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumnName, max, chunkSize);
259+
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumn, max, chunkSize);
262260
}
263261
// add the ending split
264262
splits.add(ChunkRange.of(chunkStart, null));
@@ -269,17 +267,17 @@ private Object nextChunkEnd(
269267
JdbcConnection jdbc,
270268
Object previousChunkEnd,
271269
TableId tableId,
272-
String splitColumnName,
270+
Column splitColumn,
273271
Object max,
274272
int chunkSize)
275273
throws SQLException {
276274
// chunk end might be null when max values are removed
277275
Object chunkEnd =
278-
queryNextChunkMax(jdbc, tableId, splitColumnName, chunkSize, previousChunkEnd);
276+
queryNextChunkMax(jdbc, tableId, splitColumn, chunkSize, previousChunkEnd);
279277
if (Objects.equals(previousChunkEnd, chunkEnd)) {
280278
// we don't allow equal chunk start and end,
281279
// should query the next one larger than chunkEnd
282-
chunkEnd = queryMin(jdbc, tableId, splitColumnName, chunkEnd);
280+
chunkEnd = queryMin(jdbc, tableId, splitColumn, chunkEnd);
283281
}
284282
if (ObjectUtils.compare(chunkEnd, max) >= 0) {
285283
return null;

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2ChunkSplitter.java‎

Lines changed: 16 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -127,16 +127,16 @@ public Collection<SnapshotSplit> generateSplits(TableId tableId) {
127127
}
128128

129129
@Override
130-
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
130+
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, Column column)
131131
throws SQLException {
132-
return Db2Utils.queryMinMax(jdbc, tableId, columnName);
132+
return Db2Utils.queryMinMax(jdbc, tableId, column.name());
133133
}
134134

135135
@Override
136136
public Object queryMin(
137-
JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
137+
JdbcConnection jdbc, TableId tableId, Column column, Object excludedLowerBound)
138138
throws SQLException {
139-
return Db2Utils.queryMin(jdbc, tableId, columnName, excludedLowerBound);
139+
return Db2Utils.queryMin(jdbc, tableId, column.name(), excludedLowerBound);
140140
}
141141

142142
@Override
@@ -152,11 +152,12 @@ public DataType fromDbzColumn(Column splitColumn) {
152152
public Object queryNextChunkMax(
153153
JdbcConnection jdbc,
154154
TableId tableId,
155-
String columnName,
155+
Column column,
156156
int chunkSize,
157157
Object includedLowerBound)
158158
throws SQLException {
159-
return Db2Utils.queryNextChunkMax(jdbc, tableId, columnName, chunkSize, includedLowerBound);
159+
return Db2Utils.queryNextChunkMax(
160+
jdbc, tableId, column.name(), chunkSize, includedLowerBound);
160161
}
161162

162163
@Override
@@ -177,8 +178,7 @@ public String buildSplitScanQuery(
177178
*/
178179
private List<ChunkRange> splitTableIntoChunks(
179180
JdbcConnection jdbc, TableId tableId, Column splitColumn) throws SQLException {
180-
final String splitColumnName = splitColumn.name();
181-
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumnName);
181+
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumn);
182182
final Object min = minMax[0];
183183
final Object max = minMax[1];
184184
if (min == null || max == null || min.equals(max)) {
@@ -205,11 +205,10 @@ private List<ChunkRange> splitTableIntoChunks(
205205
return splitEvenlySizedChunks(
206206
tableId, min, max, approximateRowCnt, chunkSize, dynamicChunkSize);
207207
} else {
208-
return splitUnevenlySizedChunks(
209-
jdbc, tableId, splitColumnName, min, max, chunkSize);
208+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
210209
}
211210
} else {
212-
return splitUnevenlySizedChunks(jdbc, tableId, splitColumnName, min, max, chunkSize);
211+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
213212
}
214213
}
215214

@@ -259,7 +258,7 @@ private List<ChunkRange> splitEvenlySizedChunks(
259258
private List<ChunkRange> splitUnevenlySizedChunks(
260259
JdbcConnection jdbc,
261260
TableId tableId,
262-
String splitColumnName,
261+
Column splitColumn,
263262
Object min,
264263
Object max,
265264
int chunkSize)
@@ -268,15 +267,15 @@ private List<ChunkRange> splitUnevenlySizedChunks(
268267
"Use unevenly-sized chunks for table {}, the chunk size is {}", tableId, chunkSize);
269268
final List<ChunkRange> splits = new ArrayList<>();
270269
Object chunkStart = null;
271-
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumnName, max, chunkSize);
270+
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumn, max, chunkSize);
272271
int count = 0;
273272
while (chunkEnd != null && ObjectUtils.compare(chunkEnd, max) <= 0) {
274273
// we start from [null, min + chunk_size) and avoid [null, min)
275274
splits.add(ChunkRange.of(chunkStart, chunkEnd));
276275
// may sleep awhile to avoid DDOS on Db2 server
277276
maySleep(count++, tableId);
278277
chunkStart = chunkEnd;
279-
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumnName, max, chunkSize);
278+
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumn, max, chunkSize);
280279
}
281280
// add the ending split
282281
splits.add(ChunkRange.of(chunkStart, null));
@@ -287,17 +286,17 @@ private Object nextChunkEnd(
287286
JdbcConnection jdbc,
288287
Object previousChunkEnd,
289288
TableId tableId,
290-
String splitColumnName,
289+
Column splitColumn,
291290
Object max,
292291
int chunkSize)
293292
throws SQLException {
294293
// chunk end might be null when max values are removed
295294
Object chunkEnd =
296-
queryNextChunkMax(jdbc, tableId, splitColumnName, chunkSize, previousChunkEnd);
295+
queryNextChunkMax(jdbc, tableId, splitColumn, chunkSize, previousChunkEnd);
297296
if (Objects.equals(previousChunkEnd, chunkEnd)) {
298297
// we don't allow equal chunk start and end,
299298
// should query the next one larger than chunkEnd
300-
chunkEnd = queryMin(jdbc, tableId, splitColumnName, chunkEnd);
299+
chunkEnd = queryMin(jdbc, tableId, splitColumn, chunkEnd);
301300
}
302301
if (ObjectUtils.compare(chunkEnd, max) >= 0) {
303302
return null;

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/assigner/splitter/OracleChunkSplitter.java‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -114,28 +114,28 @@ public Collection<SnapshotSplit> generateSplits(TableId tableId) {
114114
}
115115

116116
@Override
117-
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
117+
public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, Column column)
118118
throws SQLException {
119-
return OracleUtils.queryMinMax(jdbc, tableId, columnName);
119+
return OracleUtils.queryMinMax(jdbc, tableId, column.name());
120120
}
121121

122122
@Override
123123
public Object queryMin(
124-
JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
124+
JdbcConnection jdbc, TableId tableId, Column column, Object excludedLowerBound)
125125
throws SQLException {
126-
return OracleUtils.queryMin(jdbc, tableId, columnName, excludedLowerBound);
126+
return OracleUtils.queryMin(jdbc, tableId, column.name(), excludedLowerBound);
127127
}
128128

129129
@Override
130130
public Object queryNextChunkMax(
131131
JdbcConnection jdbc,
132132
TableId tableId,
133-
String columnName,
133+
Column column,
134134
int chunkSize,
135135
Object includedLowerBound)
136136
throws SQLException {
137137
return OracleUtils.queryNextChunkMax(
138-
jdbc, tableId, columnName, chunkSize, includedLowerBound);
138+
jdbc, tableId, column.name(), chunkSize, includedLowerBound);
139139
}
140140

141141
@Override
@@ -165,8 +165,7 @@ public DataType fromDbzColumn(Column splitColumn) {
165165
*/
166166
private List<ChunkRange> splitTableIntoChunks(
167167
JdbcConnection jdbc, TableId tableId, Column splitColumn) throws SQLException {
168-
final String splitColumnName = splitColumn.name();
169-
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumnName);
168+
final Object[] minMax = queryMinMax(jdbc, tableId, splitColumn);
170169
final Object min = minMax[0];
171170
final Object max = minMax[1];
172171
if (min == null || max == null || min.equals(max)) {
@@ -180,7 +179,7 @@ private List<ChunkRange> splitTableIntoChunks(
180179

181180
// use ROWID get splitUnevenlySizedChunks by default
182181
if (splitColumn.name().equals(ROWID.class.getSimpleName())) {
183-
return splitUnevenlySizedChunks(jdbc, tableId, splitColumnName, min, max, chunkSize);
182+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
184183
}
185184

186185
if (isEvenlySplitColumn(splitColumn)) {
@@ -198,11 +197,10 @@ private List<ChunkRange> splitTableIntoChunks(
198197
return splitEvenlySizedChunks(
199198
tableId, min, max, approximateRowCnt, dynamicChunkSize);
200199
} else {
201-
return splitUnevenlySizedChunks(
202-
jdbc, tableId, splitColumnName, min, max, chunkSize);
200+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
203201
}
204202
} else {
205-
return splitUnevenlySizedChunks(jdbc, tableId, splitColumnName, min, max, chunkSize);
203+
return splitUnevenlySizedChunks(jdbc, tableId, splitColumn, min, max, chunkSize);
206204
}
207205
}
208206

@@ -239,7 +237,7 @@ private List<ChunkRange> splitEvenlySizedChunks(
239237
private List<ChunkRange> splitUnevenlySizedChunks(
240238
JdbcConnection jdbc,
241239
TableId tableId,
242-
String splitColumnName,
240+
Column splitColumn,
243241
Object min,
244242
Object max,
245243
int chunkSize)
@@ -248,7 +246,7 @@ private List<ChunkRange> splitUnevenlySizedChunks(
248246
"Use unevenly-sized chunks for table {}, the chunk size is {}", tableId, chunkSize);
249247
final List<ChunkRange> splits = new ArrayList<>();
250248
Object chunkStart = null;
251-
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumnName, max, chunkSize);
249+
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumn, max, chunkSize);
252250
int count = 0;
253251

254252
while (chunkEnd != null && isChunkEndLeMax(chunkEnd, max)) {
@@ -257,7 +255,7 @@ private List<ChunkRange> splitUnevenlySizedChunks(
257255
// may sleep a while to avoid DDOS on MySQL server
258256
maySleep(count++, tableId);
259257
chunkStart = chunkEnd;
260-
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumnName, max, chunkSize);
258+
chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumn, max, chunkSize);
261259
}
262260
// add the ending split
263261
splits.add(ChunkRange.of(chunkStart, null));
@@ -294,17 +292,17 @@ private Object nextChunkEnd(
294292
JdbcConnection jdbc,
295293
Object previousChunkEnd,
296294
TableId tableId,
297-
String splitColumnName,
295+
Column splitColumn,
298296
Object max,
299297
int chunkSize)
300298
throws SQLException {
301299
// chunk end might be null when max values are removed
302300
Object chunkEnd =
303-
queryNextChunkMax(jdbc, tableId, splitColumnName, chunkSize, previousChunkEnd);
301+
queryNextChunkMax(jdbc, tableId, splitColumn, chunkSize, previousChunkEnd);
304302
if (Objects.equals(previousChunkEnd, chunkEnd)) {
305303
// we don't allow equal chunk start and end,
306304
// should query the next one larger than chunkEnd
307-
chunkEnd = queryMin(jdbc, tableId, splitColumnName, chunkEnd);
305+
chunkEnd = queryMin(jdbc, tableId, splitColumn, chunkEnd);
308306
}
309307
if (isChunkEndGeMax(chunkEnd, max)) {
310308
return null;

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ limitations under the License.
7373
<!-- fix CVE-2022-26520 https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-26520 -->
7474
<groupId>org.postgresql</groupId>
7575
<artifactId>postgresql</artifactId>
76-
<version>42.5.1</version>
76+
<version>42.7.3</version>
7777
</dependency>
7878

7979
<!-- test dependencies on Debezium -->

0 commit comments

Comments
 (0)