Skip to content

Commit a9b6e1e

Browse files
jsingh-yelpjsingh-yelp
committed
resolve llm comments
1 parent 8aac15a commit a9b6e1e

4 files changed

Lines changed: 102 additions & 115 deletions

File tree

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java‎

Lines changed: 13 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,7 @@
2929
import org.apache.flink.cdc.common.schema.Selectors;
3030
import org.apache.flink.cdc.common.source.DataSource;
3131
import org.apache.flink.cdc.common.utils.StringUtils;
32-
import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
3332
import org.apache.flink.cdc.connectors.mysql.source.MySqlDataSource;
34-
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
3533
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
3634
import org.apache.flink.cdc.connectors.mysql.source.config.ServerIdRange;
3735
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
@@ -45,7 +43,6 @@
4543
import org.apache.flink.table.catalog.ObjectPath;
4644

4745
import com.mysql.cj.conf.PropertyKey;
48-
import io.debezium.connector.mysql.MySqlConnection;
4946
import io.debezium.relational.RelationalDatabaseConnectorConfig;
5047
import org.slf4j.Logger;
5148
import org.slf4j.LoggerFactory;
@@ -58,7 +55,6 @@
5855
import java.util.Arrays;
5956
import java.util.HashMap;
6057
import java.util.HashSet;
61-
import java.util.LinkedHashMap;
6258
import java.util.List;
6359
import java.util.Map;
6460
import java.util.Optional;
@@ -230,11 +226,15 @@ public DataSource createDataSource(Context context) {
230226

231227
// Resolve concrete table names for lineage (always, regardless of binlog mode)
232228
Selectors tableSelectors = new Selectors.SelectorsBuilder().includeTables(tables).build();
233-
List<String> tableList = getTableList(tableIds, tableSelectors);
229+
List<String> capturedTables = getTableList(tableIds, tableSelectors);
230+
boolean hasIncludedTables = !capturedTables.isEmpty();
234231
if (tablesExclude != null) {
235-
Selectors excludeSelectors =
232+
Selectors selectExclude =
236233
new Selectors.SelectorsBuilder().includeTables(tablesExclude).build();
237-
tableList.removeAll(getTableList(tableIds, excludeSelectors));
234+
List<String> excludeTables = getTableList(tableIds, selectExclude);
235+
if (!excludeTables.isEmpty()) {
236+
capturedTables.removeAll(excludeTables);
237+
}
238238
}
239239

240240
if (scanBinlogNewlyAddedTableEnabled && scanNewlyAddedTableEnabled) {
@@ -248,24 +248,14 @@ public DataSource createDataSource(Context context) {
248248
configFactory.excludeTableList(tablesExclude);
249249

250250
} else {
251-
Selectors selectors = new Selectors.SelectorsBuilder().includeTables(tables).build();
252-
List<String> capturedTables = getTableList(tableIds, selectors);
253-
if (capturedTables.isEmpty()) {
251+
if (!hasIncludedTables) {
254252
throw new IllegalArgumentException(
255253
"Cannot find any table by the option 'tables' = " + tables);
256254
}
257-
if (tablesExclude != null) {
258-
Selectors selectExclude =
259-
new Selectors.SelectorsBuilder().includeTables(tablesExclude).build();
260-
List<String> excludeTables = getTableList(tableIds, selectExclude);
261-
if (!excludeTables.isEmpty()) {
262-
capturedTables.removeAll(excludeTables);
263-
}
264-
if (capturedTables.isEmpty()) {
265-
throw new IllegalArgumentException(
266-
"Cannot find any table with by the option 'tables.exclude' = "
267-
+ tablesExclude);
268-
}
255+
if (capturedTables.isEmpty()) {
256+
throw new IllegalArgumentException(
257+
"Cannot find any table with by the option 'tables.exclude' = "
258+
+ tablesExclude);
269259
}
270260
configFactory.tableList(capturedTables.toArray(new String[0]));
271261
}
@@ -297,13 +287,9 @@ public DataSource createDataSource(Context context) {
297287
LOG.info("Add chunkKeyColumn {}.", chunkKeyColumnMap);
298288
configFactory.chunkKeyColumn(chunkKeyColumnMap);
299289
}
300-
// Fetch table schemas for lineage.
301-
Map<String, LinkedHashMap<String, String>> tableSchemas =
302-
fetchTableSchemas(configFactory, tableList);
303-
304290
String metadataList = config.get(METADATA_LIST);
305291
List<MySqlReadableMetadata> readableMetadataList = listReadableMetadata(metadataList);
306-
return new MySqlDataSource(configFactory, readableMetadataList, tableList, tableSchemas);
292+
return new MySqlDataSource(configFactory, readableMetadataList, capturedTables);
307293
}
308294

309295
private List<MySqlReadableMetadata> listReadableMetadata(String metadataList) {
@@ -544,49 +530,4 @@ private static ZoneId getServerTimeZone(Configuration config) {
544530
return ZoneId.systemDefault();
545531
}
546532
}
547-
548-
/** Fetch table schemas for lineage schema facets. */
549-
private static Map<String, LinkedHashMap<String, String>> fetchTableSchemas(
550-
MySqlSourceConfigFactory configFactory, List<String> tableList) {
551-
Map<String, LinkedHashMap<String, String>> schemas = new HashMap<>();
552-
if (tableList == null || tableList.isEmpty()) {
553-
return schemas;
554-
}
555-
try {
556-
MySqlSourceConfig sourceConfig = configFactory.createConfig(0);
557-
try (MySqlConnection jdbc = DebeziumUtils.createMySqlConnection(sourceConfig)) {
558-
for (String table : tableList) {
559-
try {
560-
TableId tableId = TableId.parse(table);
561-
LinkedHashMap<String, String> fields = new LinkedHashMap<>();
562-
jdbc.query(
563-
"SHOW COLUMNS FROM "
564-
+ quoteIdentifier(tableId.getTableName())
565-
+ " FROM "
566-
+ quoteIdentifier(tableId.getSchemaName()),
567-
rs -> {
568-
while (rs.next()) {
569-
String name = rs.getString("Field");
570-
String type = rs.getString("Type");
571-
String nullable = rs.getString("Null");
572-
fields.put(
573-
name,
574-
"YES".equals(nullable) ? type : type + " NOT NULL");
575-
}
576-
});
577-
schemas.put(table, fields);
578-
} catch (Exception e) {
579-
LOG.warn("Failed to fetch schema for table {}: {}", table, e.getMessage());
580-
}
581-
}
582-
}
583-
} catch (Exception e) {
584-
LOG.warn("Failed to fetch table schemas for lineage: {}", e.getMessage());
585-
}
586-
return schemas;
587-
}
588-
589-
private static String quoteIdentifier(String identifier) {
590-
return "`" + identifier.replace("`", "``") + "`";
591-
}
592533
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -35,9 +35,7 @@
3535
import io.debezium.relational.RelationalDatabaseConnectorConfig;
3636

3737
import java.util.ArrayList;
38-
import java.util.LinkedHashMap;
3938
import java.util.List;
40-
import java.util.Map;
4139

4240
/** A {@link DataSource} for mysql cdc connector. */
4341
@Internal
@@ -46,30 +44,27 @@ public class MySqlDataSource implements DataSource {
4644
private final MySqlSourceConfigFactory configFactory;
4745
private final MySqlSourceConfig sourceConfig;
4846
private final List<String> tableList;
49-
private final Map<String, LinkedHashMap<String, String>> tableSchemas;
5047

5148
private List<MySqlReadableMetadata> readableMetadataList;
5249

5350
public MySqlDataSource(MySqlSourceConfigFactory configFactory) {
54-
this(configFactory, new ArrayList<>(), null, null);
51+
this(configFactory, new ArrayList<>(), null);
5552
}
5653

5754
public MySqlDataSource(
5855
MySqlSourceConfigFactory configFactory,
5956
List<MySqlReadableMetadata> readableMetadataList) {
60-
this(configFactory, readableMetadataList, null, null);
57+
this(configFactory, readableMetadataList, null);
6158
}
6259

6360
public MySqlDataSource(
6461
MySqlSourceConfigFactory configFactory,
6562
List<MySqlReadableMetadata> readableMetadataList,
66-
List<String> tableList,
67-
Map<String, LinkedHashMap<String, String>> tableSchemas) {
63+
List<String> tableList) {
6864
this.configFactory = configFactory;
6965
this.sourceConfig = configFactory.createConfig(0);
7066
this.readableMetadataList = readableMetadataList;
7167
this.tableList = tableList;
72-
this.tableSchemas = tableSchemas;
7368
}
7469

7570
@Override
@@ -100,8 +95,7 @@ public EventSourceProvider getEventSourceProvider() {
10095
sourceReaderMetrics,
10196
sourceConfig,
10297
isTableIdCaseInsensitive),
103-
tableList,
104-
tableSchemas);
98+
tableList);
10599

106100
return FlinkSourceProvider.of(source);
107101
}

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java‎

Lines changed: 80 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.flink.cdc.common.annotation.Internal;
2929
import org.apache.flink.cdc.common.annotation.PublicEvolving;
3030
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
31+
import org.apache.flink.cdc.common.event.TableId;
3132
import org.apache.flink.cdc.common.lineage.LineageUtils;
3233
import org.apache.flink.cdc.connectors.mysql.MySqlValidator;
3334
import org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils;
@@ -59,15 +60,20 @@
5960
import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
6061
import org.apache.flink.util.FlinkRuntimeException;
6162

63+
import io.debezium.connector.mysql.MySqlConnection;
6264
import io.debezium.jdbc.JdbcConnection;
65+
import org.slf4j.Logger;
66+
import org.slf4j.LoggerFactory;
6367

6468
import java.io.Serializable;
6569
import java.lang.reflect.Method;
6670
import java.util.ArrayList;
71+
import java.util.HashMap;
6772
import java.util.LinkedHashMap;
6873
import java.util.List;
6974
import java.util.Map;
7075
import java.util.function.Supplier;
76+
import java.util.stream.Collectors;
7177

7278
/**
7379
* The MySQL CDC Source based on FLIP-27 and Watermark Signal Algorithm which supports parallel
@@ -105,13 +111,15 @@ public class MySqlSource<T>
105111

106112
private static final long serialVersionUID = 1L;
107113

114+
private static final Logger LOG = LoggerFactory.getLogger(MySqlSource.class);
115+
108116
private static final String ENUMERATOR_SERVER_NAME = "mysql_source_split_enumerator";
109117

110118
private final MySqlSourceConfigFactory configFactory;
111119
private final DebeziumDeserializationSchema<T> deserializationSchema;
112120
private final RecordEmitterSupplier<T> recordEmitterSupplier;
113121
private final List<String> tableList;
114-
private final Map<String, LinkedHashMap<String, String>> tableSchemas;
122+
private transient Map<String, LinkedHashMap<String, String>> fetchedLineageTableSchemas;
115123

116124
// Actions to perform during the snapshot phase.
117125
// This field is introduced for testing purpose, for example testing if changes made in the
@@ -148,20 +156,18 @@ public static <T> MySqlSourceBuilder<T> builder() {
148156
MySqlSourceConfigFactory configFactory,
149157
DebeziumDeserializationSchema<T> deserializationSchema,
150158
RecordEmitterSupplier<T> recordEmitterSupplier) {
151-
this(configFactory, deserializationSchema, recordEmitterSupplier, null, null);
159+
this(configFactory, deserializationSchema, recordEmitterSupplier, null);
152160
}
153161

154162
MySqlSource(
155163
MySqlSourceConfigFactory configFactory,
156164
DebeziumDeserializationSchema<T> deserializationSchema,
157165
RecordEmitterSupplier<T> recordEmitterSupplier,
158-
List<String> tableList,
159-
Map<String, LinkedHashMap<String, String>> tableSchemas) {
166+
List<String> tableList) {
160167
this.configFactory = configFactory;
161168
this.deserializationSchema = deserializationSchema;
162169
this.recordEmitterSupplier = recordEmitterSupplier;
163-
this.tableList = tableList;
164-
this.tableSchemas = tableSchemas;
170+
this.tableList = tableList == null ? null : new ArrayList<>(tableList);
165171
}
166172

167173
public MySqlSourceConfigFactory getConfigFactory() {
@@ -181,13 +187,80 @@ public Boundedness getBoundedness() {
181187
@Override
182188
public LineageVertex getLineageVertex() {
183189
MySqlSourceConfig sourceConfig = configFactory.createConfig(0);
190+
191+
/*
192+
* Note: Flink collects lineage during job graph construction, so this includes tables that
193+
* already exist when the job is submitted. If `scan.binlog.newly-added-table.enabled` later
194+
* captures newly created tables while the job is running, those tables appear in lineage
195+
* after the next submission.
196+
*/
184197
return LineageUtils.sourceLineageVertex(
185198
"mysql",
186199
sourceConfig.getHostname(),
187200
sourceConfig.getPort(),
188201
sourceConfig.getStartupOptions().isSnapshotOnly(),
189202
tableList,
190-
tableSchemas);
203+
fetchLineageTableSchemas(sourceConfig));
204+
}
205+
206+
/** Fetch table schemas for lineage schema facets, only called when lineage is enabled. */
207+
private synchronized Map<String, LinkedHashMap<String, String>> fetchLineageTableSchemas(
208+
MySqlSourceConfig sourceConfig) {
209+
if (fetchedLineageTableSchemas != null) {
210+
return fetchedLineageTableSchemas;
211+
}
212+
Map<String, LinkedHashMap<String, String>> schemas = new HashMap<>();
213+
if (tableList == null || tableList.isEmpty()) {
214+
fetchedLineageTableSchemas = schemas;
215+
return fetchedLineageTableSchemas;
216+
}
217+
List<TableId> tableIds =
218+
tableList.stream().map(TableId::parse).collect(Collectors.toList());
219+
tableList.forEach(table -> schemas.put(table, new LinkedHashMap<>()));
220+
try (MySqlConnection jdbc = DebeziumUtils.createMySqlConnection(sourceConfig)) {
221+
jdbc.query(
222+
"SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE "
223+
+ "FROM information_schema.COLUMNS WHERE "
224+
+ tableFilter(tableIds)
225+
+ " ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION",
226+
rs -> {
227+
while (rs.next()) {
228+
String table =
229+
TableId.tableId(
230+
rs.getString("TABLE_SCHEMA"),
231+
rs.getString("TABLE_NAME"))
232+
.toString();
233+
LinkedHashMap<String, String> fields = schemas.get(table);
234+
if (fields != null) {
235+
String type = rs.getString("COLUMN_TYPE");
236+
String nullable = rs.getString("IS_NULLABLE");
237+
fields.put(
238+
rs.getString("COLUMN_NAME"),
239+
"YES".equals(nullable) ? type : type + " NOT NULL");
240+
}
241+
}
242+
});
243+
} catch (Exception e) {
244+
LOG.warn("Failed to fetch table schemas for lineage: {}", e.getMessage());
245+
}
246+
fetchedLineageTableSchemas = schemas;
247+
return fetchedLineageTableSchemas;
248+
}
249+
250+
private static String tableFilter(List<TableId> tableIds) {
251+
return tableIds.stream()
252+
.map(
253+
tableId ->
254+
"(TABLE_SCHEMA = "
255+
+ quoteStringLiteral(tableId.getSchemaName())
256+
+ " AND TABLE_NAME = "
257+
+ quoteStringLiteral(tableId.getTableName())
258+
+ ")")
259+
.collect(Collectors.joining(" OR "));
260+
}
261+
262+
private static String quoteStringLiteral(String literal) {
263+
return "'" + literal.replace("'", "''") + "'";
191264
}
192265

193266
@Override

0 commit comments

Comments
 (0)