github-actions[bot] commented on code in PR #65325:
URL: https://github.com/apache/doris/pull/65325#discussion_r3550201584
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
+ case "SMALLINT":
+ case "INT2":
+ case "YEAR":
+ return unsigned ? DorisType.INT : DorisType.SMALLINT;
+ case "MEDIUMINT":
+ case "INT":
+ case "INTEGER":
+ case "INT3":
+ case "INT4":
+ return unsigned ? DorisType.BIGINT : DorisType.INT;
+ case "BIGINT":
+ case "INT8":
+ return unsigned ? DorisType.LARGEINT : DorisType.BIGINT;
+ case "FLOAT":
+ case "FLOAT4":
+ return DorisType.FLOAT;
+ case "DOUBLE":
+ case "FLOAT8":
+ case "REAL":
+ return DorisType.DOUBLE;
+ case "DECIMAL":
+ case "DEC":
+ case "FIXED":
+ case "NUMERIC":
+ {
+ int precision = length > 0 ? length : 10;
+ if (unsigned) {
+ precision++;
+ }
+ if (precision > MAX_DECIMAL128_PRECISION) {
+ return DorisType.STRING;
+ }
+ int decimalScale = scale >= 0 ? scale : 0;
+ return String.format("%s(%d, %d)", DorisType.DECIMAL,
precision, decimalScale);
+ }
+ case "DATE":
+ return DorisType.DATE;
+ case "DATETIME":
+ case "TIMESTAMP":
+ {
+ int timeScale = mysqlTimeScale(length, scale);
+ return String.format("%s(%d)", DorisType.DATETIME,
timeScale);
+ }
+ case "CHAR":
+ case "NCHAR":
+ return charToDorisType(length);
+ case "VARCHAR":
+ case "NVARCHAR":
+ return varcharToDorisType(length);
+ case "BIT":
+ return length == 1 ? DorisType.BOOLEAN : DorisType.STRING;
+ case "JSON":
+ return DorisType.JSON;
Review Comment:
This MySQL ADD COLUMN mapper still diverges from initial table creation for
JSON. When the column exists at job creation, FE goes through
`JdbcMySQLClient.jdbcTypeToDoris()` and maps MySQL `JSON` to `STRING`, but a
later `ALTER TABLE ... ADD COLUMN payload JSON` reaches this new branch and
emits Doris `JSON`. That means the same upstream schema gets a different Doris
type depending on whether the column existed before the job started or was
added later. Please mirror the initial MySQL JDBC mapping here, or change both
paths together with coverage for the intended JSON behavior.
##########
fe/fe-common/src/main/java/org/apache/doris/job/cdc/DataSourceConfigKeys.java:
##########
@@ -39,6 +39,7 @@ public class DataSourceConfigKeys {
public static final String SNAPSHOT_PARALLELISM = "snapshot_parallelism";
public static final String SNAPSHOT_PARALLELISM_DEFAULT = "1";
public static final String SKIP_SNAPSHOT_BACKFILL =
"skip_snapshot_backfill";
+ public static final String SCHEMA_CHANGE_ENABLED = "schema_change_enabled";
Review Comment:
Please also register this new source property in FE validation. The readers
now consume `schema_change_enabled` and tests cover setting it to `false`, but
`CreateJobCommand`/`AlterJobCommand` still pass source properties through
`DataSourceConfigValidator.validateSource()`, whose `ALLOW_SOURCE_KEYS` does
not include this key. A user trying `CREATE JOB ... FROM MYSQL/POSTGRES
("schema_change_enabled"="false", ...)` will get `Unexpected key:
'schema_change_enabled'` before the request reaches cdc_client, so the new
disable path is only usable from internally injected TVF properties, not from
the streaming job source config.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/MySqlDebeziumJsonDeserializer.java:
##########
@@ -53,14 +69,185 @@ public void init(Map<String, String> props) {
public DeserializeResult deserialize(Map<String, String> context,
SourceRecord record)
throws IOException {
if (RecordUtils.isSchemaChangeEvent(record)) {
+ if (!isSchemaChangeEnabled(context)) {
+ LOG.info("[SCHEMA-CHANGE-SKIPPED] MySQL schema change is
disabled.");
+ return DeserializeResult.empty();
+ }
return handleSchemaChangeEvent(record, context);
}
return super.deserialize(context, record);
}
private DeserializeResult handleSchemaChangeEvent(
- SourceRecord record, Map<String, String> context) {
- // todo: record has mysql ddl, need to convert doris ddl
- return DeserializeResult.empty();
+ SourceRecord record, Map<String, String> context) throws
IOException {
+ HistoryRecord history = RecordUtils.getHistoryRecord(record);
+ String database =
history.document().getString(HistoryRecord.Fields.DATABASE_NAME);
+ String ddl =
history.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
+ Map<TableId, TableChanges.TableChange> freshSchemas =
deserializeTableChanges(history);
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL deserializer received schema change,
baselineSchemas={}, freshSchemas={}, ddl={}",
+ tableSchemas == null ? 0 : tableSchemas.size(),
+ freshSchemas.size(),
+ ddl);
+ if (freshSchemas.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event has no
table schema "
+ + "change. DDL: {}",
+ ddl);
+ return DeserializeResult.empty();
+ }
+
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL schema baseline is empty, adopting
fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL: {}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (TableId tableId : freshSchemas.keySet()) {
+ if (!tableSchemas.containsKey(tableId)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {} has no baseline,
adopting fresh schemas as "
+ + "baseline without emitting Doris DDL. DDL:
{}",
+ tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ Tables tables = toTables(tableSchemas);
+ List<MySqlSchemaChange> changes = Collections.emptyList();
+ try {
+ CustomMySqlAntlrDdlParser parser = new CustomMySqlAntlrDdlParser();
+ parser.setCurrentDatabase(database);
+ parser.parse(ddl, tables);
+ changes = parser.getAndClearParsedChanges();
+ } catch (Exception e) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change DDL parser
failed. No Doris DDL "
+ + "emitted, FE baseline will advance to the schema
carried by the "
+ + "history record. DDL: {}",
+ ddl,
+ e);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ if (changes.isEmpty()) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change event
produced no supported "
+ + "column change. No Doris DDL emitted, FE
baseline will advance to "
+ + "the schema carried by the history record. DDL:
{}",
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+
+ List<MySqlSchemaChange> supportedChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : changes) {
+ if (change.getType() == MySqlSchemaChange.Type.UNSUPPORTED) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL unsupported schema
change {} for table {}: {}."
+ + " Doris DDL will not be emitted for this
change, FE baseline will"
+ + " advance to the schema carried by the
history record. DDL: {}",
+ change.getSourceOperation(),
+ change.getTableId() == null
+ ? "<unknown>"
+ : change.getTableId().identifier(),
+ change.getReason(),
+ ddl);
+ continue;
+ }
+ supportedChanges.add(change);
+ }
+ if (supportedChanges.isEmpty()) {
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ if (tableId == null || !tableSchemas.containsKey(tableId)) {
+ LOG.warn(
+ "[SCHEMA-CHANGE-SKIPPED] MySQL schema change table {}
has no baseline."
+ + " No Doris DDL emitted, FE baseline will
advance to the schema"
+ + " carried by the history record. DDL: {}",
+ tableId == null ? "<unknown>" : tableId.identifier(),
+ ddl);
+ return DeserializeResult.schemaChange(
+ Collections.emptyList(), freshSchemas,
Collections.emptyList());
+ }
+ }
+
+ List<SchemaChangeOperation> schemaChanges = new ArrayList<>();
+ for (MySqlSchemaChange change : supportedChanges) {
+ TableId tableId = change.getTableId();
+ Set<String> excludedCols =
+ excludeColumnsCache.getOrDefault(tableId.table(),
Collections.emptySet());
+ String targetTable = resolveTargetTable(tableId.table());
+ String db = context.get(Constants.DORIS_TARGET_DB);
+ if (change.getType() == MySqlSchemaChange.Type.ADD) {
+ String columnName = change.getColumn().name();
+ if (excludedCols.contains(columnName)) {
+ LOG.info(
+ "[SCHEMA-CHANGE] MySQL table {}: added column '{}'
is excluded, skipping ADD",
+ tableId.identifier(),
+ columnName);
+ continue;
+ }
+ String colType =
SchemaChangeHelper.mysqlColumnToDorisType(change.getColumn());
+ schemaChanges.add(
+ SchemaChangeOperation.addColumn(
+ targetTable,
+ columnName,
+ SchemaChangeHelper.buildAddColumnSql(
Review Comment:
This generated ADD COLUMN DDL drops MySQL default/backfill and `NOT NULL`
semantics. For example, `ALTER TABLE ... ADD COLUMN c INT NOT NULL DEFAULT 1`
backfills existing rows in MySQL and records the column as non-null, but this
path emits only `ALTER TABLE ... ADD COLUMN c INT` plus an optional comment, so
existing Doris rows remain `NULL` and the target column is nullable. Initial
table creation does preserve JDBC nullability through
`getColumnsFromJdbc()`/`ColumnDefinition`, so the same source column behaves
differently depending on when it is introduced. If the intended schema-change
contract is to keep added columns nullable to avoid backfill, please
detect/default such MySQL ADDs and avoid treating them as fully applied;
otherwise carry the default/backfill/nullability into the Doris DDL.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
+ case "SMALLINT":
+ case "INT2":
+ case "YEAR":
+ return unsigned ? DorisType.INT : DorisType.SMALLINT;
+ case "MEDIUMINT":
+ case "INT":
+ case "INTEGER":
+ case "INT3":
+ case "INT4":
+ return unsigned ? DorisType.BIGINT : DorisType.INT;
+ case "BIGINT":
+ case "INT8":
+ return unsigned ? DorisType.LARGEINT : DorisType.BIGINT;
+ case "FLOAT":
+ case "FLOAT4":
+ return DorisType.FLOAT;
+ case "DOUBLE":
+ case "FLOAT8":
+ case "REAL":
+ return DorisType.DOUBLE;
+ case "DECIMAL":
+ case "DEC":
+ case "FIXED":
+ case "NUMERIC":
+ {
+ int precision = length > 0 ? length : 10;
+ if (unsigned) {
+ precision++;
+ }
+ if (precision > MAX_DECIMAL128_PRECISION) {
+ return DorisType.STRING;
+ }
+ int decimalScale = scale >= 0 ? scale : 0;
+ return String.format("%s(%d, %d)", DorisType.DECIMAL,
precision, decimalScale);
+ }
+ case "DATE":
+ return DorisType.DATE;
+ case "DATETIME":
+ case "TIMESTAMP":
+ {
+ int timeScale = mysqlTimeScale(length, scale);
+ return String.format("%s(%d)", DorisType.DATETIME,
timeScale);
+ }
+ case "CHAR":
+ case "NCHAR":
+ return charToDorisType(length);
+ case "VARCHAR":
+ case "NVARCHAR":
+ return varcharToDorisType(length);
+ case "BIT":
+ return length == 1 ? DorisType.BOOLEAN : DorisType.STRING;
+ case "JSON":
+ return DorisType.JSON;
+ case "TINYBLOB":
+ case "BLOB":
+ case "MEDIUMBLOB":
+ case "LONGBLOB":
+ case "BINARY":
+ case "VARBINARY":
+ case "TIME":
+ case "TINYTEXT":
+ case "TEXT":
+ case "MEDIUMTEXT":
+ case "LONGTEXT":
+ case "STRING":
+ case "SET":
+ case "ENUM":
+ return DorisType.STRING;
+ default:
+ LOG.warn("Unrecognized MySQL type '{}', defaulting to STRING",
mysqlTypeName);
+ return DorisType.STRING;
+ }
+ }
+
+ private static int mysqlTimeScale(int length, int scale) {
+ if (scale >= 0 && scale <= 6) {
+ return scale;
+ }
+ if (length >= 0 && length <= 6) {
+ return length;
+ }
+ return 0;
+ }
+
+ private static String charToDorisType(int length) {
+ if (length <= 0) {
+ return DorisType.STRING;
+ }
+ int len = length * 3;
+ if (len > MAX_VARCHAR_LENGTH) {
+ return DorisType.STRING;
+ }
+ if (len > MAX_CHAR_LENGTH) {
+ return String.format("%s(%d)", DorisType.VARCHAR, len);
+ }
+ return String.format("%s(%d)", DorisType.CHAR, len);
+ }
+
+ private static String varcharToDorisType(int length) {
+ if (length <= 0) {
+ return DorisType.STRING;
+ }
+ int len = length * 3;
+ if (len > MAX_VARCHAR_LENGTH) {
+ return DorisType.STRING;
+ }
+ return String.format("%s(%d)", DorisType.VARCHAR, len);
+ }
+
/** Map a PostgreSQL native type name to a Doris type string. */
static String pgTypeNameToDorisType(String pgTypeName, int length, int
scale) {
Review Comment:
Please do not default unrecognized PostgreSQL ADD COLUMN types to Doris
`STRING`. Initial table creation goes through
`JdbcPostgreSQLClient.jdbcTypeToDoris()`, which returns `UNSUPPORTED` for
unknown non-array source types, and `StreamingJobUtils.getColumns()` rejects
those columns before creating the target table. This schema-change path instead
maps the same unknown type to `STRING`, so a source column can be rejected when
present at job creation but silently accepted after startup, with the FE
baseline advanced as if the source DDL were faithfully applied. Please align
this path with the JDBC mapper, or add an intentional shared mapping with
coverage.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
+ case "SMALLINT":
+ case "INT2":
+ case "YEAR":
+ return unsigned ? DorisType.INT : DorisType.SMALLINT;
+ case "MEDIUMINT":
Review Comment:
This branch over-widens `MEDIUMINT UNSIGNED` compared with initial table
creation. `JdbcMySQLClient.jdbcTypeToDoris()` maps both `SMALLINT UNSIGNED` and
`MEDIUMINT UNSIGNED` to Doris `INT`, and the existing integer-boundary expected
output records `mediumint_u` as `int`. Here `MEDIUMINT` shares the `INT`
branch, so once `UNSIGNED` is detected it returns Doris `BIGINT`. An upstream
`MEDIUMINT UNSIGNED` column therefore has a different Doris type depending on
whether it existed before the job started or was added later. Please keep this
mapper aligned with the JDBC path, while still mapping `INT UNSIGNED` to
`BIGINT`.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
Review Comment:
This PostgreSQL ADD COLUMN mapper diverges from initial table creation for
`json`/`jsonb`. The JDBC creation path maps those source types to Doris
string/text, and existing generated baselines record PostgreSQL `json` and
`jsonb` columns as `text`; this schema-change path returns Doris `JSON`
instead. That means the same upstream column has a different Doris type
depending on whether it existed before the job started or was added later.
Please align this mapper with `JdbcPostgreSQLClient`, or change both paths
together with regression coverage for the intended PostgreSQL JSON behavior.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
+ case "SMALLINT":
+ case "INT2":
+ case "YEAR":
+ return unsigned ? DorisType.INT : DorisType.SMALLINT;
+ case "MEDIUMINT":
+ case "INT":
+ case "INTEGER":
+ case "INT3":
+ case "INT4":
+ return unsigned ? DorisType.BIGINT : DorisType.INT;
+ case "BIGINT":
+ case "INT8":
+ return unsigned ? DorisType.LARGEINT : DorisType.BIGINT;
+ case "FLOAT":
+ case "FLOAT4":
+ return DorisType.FLOAT;
+ case "DOUBLE":
+ case "FLOAT8":
+ case "REAL":
+ return DorisType.DOUBLE;
+ case "DECIMAL":
+ case "DEC":
+ case "FIXED":
+ case "NUMERIC":
+ {
+ int precision = length > 0 ? length : 10;
+ if (unsigned) {
+ precision++;
+ }
+ if (precision > MAX_DECIMAL128_PRECISION) {
+ return DorisType.STRING;
+ }
+ int decimalScale = scale >= 0 ? scale : 0;
+ return String.format("%s(%d, %d)", DorisType.DECIMAL,
precision, decimalScale);
+ }
+ case "DATE":
+ return DorisType.DATE;
+ case "DATETIME":
+ case "TIMESTAMP":
+ {
+ int timeScale = mysqlTimeScale(length, scale);
+ return String.format("%s(%d)", DorisType.DATETIME,
timeScale);
+ }
+ case "CHAR":
+ case "NCHAR":
+ return charToDorisType(length);
+ case "VARCHAR":
+ case "NVARCHAR":
+ return varcharToDorisType(length);
+ case "BIT":
+ return length == 1 ? DorisType.BOOLEAN : DorisType.STRING;
+ case "JSON":
+ return DorisType.JSON;
+ case "TINYBLOB":
+ case "BLOB":
+ case "MEDIUMBLOB":
+ case "LONGBLOB":
+ case "BINARY":
+ case "VARBINARY":
+ case "TIME":
+ case "TINYTEXT":
+ case "TEXT":
+ case "MEDIUMTEXT":
+ case "LONGTEXT":
+ case "STRING":
+ case "SET":
+ case "ENUM":
+ return DorisType.STRING;
+ default:
+ LOG.warn("Unrecognized MySQL type '{}', defaulting to STRING",
mysqlTypeName);
+ return DorisType.STRING;
+ }
+ }
+
+ private static int mysqlTimeScale(int length, int scale) {
+ if (scale >= 0 && scale <= 6) {
+ return scale;
+ }
+ if (length >= 0 && length <= 6) {
+ return length;
+ }
+ return 0;
+ }
+
+ private static String charToDorisType(int length) {
+ if (length <= 0) {
+ return DorisType.STRING;
+ }
+ int len = length * 3;
+ if (len > MAX_VARCHAR_LENGTH) {
+ return DorisType.STRING;
+ }
+ if (len > MAX_CHAR_LENGTH) {
+ return String.format("%s(%d)", DorisType.VARCHAR, len);
+ }
+ return String.format("%s(%d)", DorisType.CHAR, len);
+ }
+
+ private static String varcharToDorisType(int length) {
+ if (length <= 0) {
+ return DorisType.STRING;
+ }
+ int len = length * 3;
+ if (len > MAX_VARCHAR_LENGTH) {
+ return DorisType.STRING;
+ }
+ return String.format("%s(%d)", DorisType.VARCHAR, len);
+ }
+
/** Map a PostgreSQL native type name to a Doris type string. */
static String pgTypeNameToDorisType(String pgTypeName, int length, int
scale) {
Preconditions.checkNotNull(pgTypeName);
Review Comment:
This array shortcut accepts PostgreSQL array types that initial table
creation rejects. The JDBC path only allows array inner types from
`supportedInnerType`; if the inner type is not in that list,
`convertArrayType()` returns `UNSUPPORTED` and the job creation path rejects
the column. Here any `_type` is recursively mapped, so arrays such as `_bytea`,
`_time`, or `_hstore` become Doris `ARRAY<STRING>` when added after startup.
Please apply the same PostgreSQL array allowlist as `JdbcPostgreSQLClient`, or
add an intentional shared mapping with coverage for the extra array types.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
Review Comment:
This branch also needs to preserve the existing MySQL `TINYINT(1)` boolean
mapping. The initial streaming-job path already expects a source `bool_col
tinyint(1)` to become Doris `boolean` (see the committed
`test_streaming_mysql_job_integer_boundary` expected `DESC` output), but an
`ALTER TABLE ... ADD COLUMN flag TINYINT(1)` reaches this mapper and returns
Doris `TINYINT` because the branch ignores `column.length()`. That makes the
same source schema produce `BOOLEAN` or `TINYINT` depending on when the column
appears. Please mirror the initial mapping here, or change both paths together
with coverage for the intended behavior.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -136,6 +109,130 @@ public static String columnToDorisType(Column column) {
return pgTypeNameToDorisType(column.typeName(), column.length(),
column.scale().orElse(-1));
}
+ /** Convert a Debezium MySQL Column to a Doris column type string. */
+ public static String mysqlColumnToDorisType(Column column) {
+ Preconditions.checkNotNull(column.typeName());
+ String mysqlTypeName = column.typeName().toUpperCase();
+ String[] typeFields = mysqlTypeName.split(" ");
+ String mysqlType = typeFields[0];
+ boolean unsigned = typeFields.length > 1 &&
"UNSIGNED".equals(typeFields[1]);
+ int length = column.length();
+ int scale = column.scale().orElse(-1);
+ switch (mysqlType) {
+ case "BOOLEAN":
+ case "BOOL":
+ return DorisType.BOOLEAN;
+ case "TINYINT":
+ return unsigned ? DorisType.SMALLINT : DorisType.TINYINT;
+ case "SMALLINT":
+ case "INT2":
+ case "YEAR":
+ return unsigned ? DorisType.INT : DorisType.SMALLINT;
+ case "MEDIUMINT":
+ case "INT":
+ case "INTEGER":
+ case "INT3":
+ case "INT4":
+ return unsigned ? DorisType.BIGINT : DorisType.INT;
+ case "BIGINT":
+ case "INT8":
+ return unsigned ? DorisType.LARGEINT : DorisType.BIGINT;
+ case "FLOAT":
+ case "FLOAT4":
+ return DorisType.FLOAT;
+ case "DOUBLE":
+ case "FLOAT8":
+ case "REAL":
+ return DorisType.DOUBLE;
+ case "DECIMAL":
+ case "DEC":
+ case "FIXED":
+ case "NUMERIC":
+ {
+ int precision = length > 0 ? length : 10;
+ if (unsigned) {
+ precision++;
+ }
+ if (precision > MAX_DECIMAL128_PRECISION) {
+ return DorisType.STRING;
+ }
+ int decimalScale = scale >= 0 ? scale : 0;
+ return String.format("%s(%d, %d)", DorisType.DECIMAL,
precision, decimalScale);
+ }
+ case "DATE":
+ return DorisType.DATE;
+ case "DATETIME":
+ case "TIMESTAMP":
+ {
+ int timeScale = mysqlTimeScale(length, scale);
+ return String.format("%s(%d)", DorisType.DATETIME,
timeScale);
+ }
+ case "CHAR":
+ case "NCHAR":
+ return charToDorisType(length);
+ case "VARCHAR":
+ case "NVARCHAR":
+ return varcharToDorisType(length);
+ case "BIT":
+ return length == 1 ? DorisType.BOOLEAN : DorisType.STRING;
+ case "JSON":
+ return DorisType.JSON;
+ case "TINYBLOB":
+ case "BLOB":
+ case "MEDIUMBLOB":
+ case "LONGBLOB":
+ case "BINARY":
+ case "VARBINARY":
+ case "TIME":
+ case "TINYTEXT":
+ case "TEXT":
+ case "MEDIUMTEXT":
+ case "LONGTEXT":
+ case "STRING":
+ case "SET":
+ case "ENUM":
+ return DorisType.STRING;
+ default:
+ LOG.warn("Unrecognized MySQL type '{}', defaulting to STRING",
mysqlTypeName);
Review Comment:
Please do not let unsupported MySQL ADD COLUMN types fall through to
`STRING` here. The custom parser registers spatial types such as
`POINT`/`GEOMETRY`, but this mapper has no matching cases, so `ALTER TABLE ...
ADD COLUMN g POINT` reaches the default branch and becomes a Doris `STRING`
column. Initial job creation goes through `JdbcMySQLClient.jdbcTypeToDoris()`
and `StreamingJobUtils.getColumns()`, which reject unsupported source columns
instead of silently remapping them. This makes the same source schema accepted
or rejected depending on when the column appears, and advances the schema
baseline as if the ADD were faithfully applied. Please reject/skip these types
without treating the DDL as applied, or add an explicit shared mapping with
coverage.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java:
##########
@@ -96,6 +100,11 @@ private DeserializeResult handleSchemaChangeEvent(
// Debezium PG TableId is (catalog=null, schema, table) — matches the
tableSchemas key.
TableId tableId = freshTable.id();
TableChanges.TableChange stored = tableSchemas != null ?
tableSchemas.get(tableId) : null;
+ LOG.info(
+ "[SCHEMA-CHANGE] Postgres deserializer received schema change,
table={}, baselineSchemas={}, hasStoredSchema={}",
+ tableId.identifier(),
+ tableSchemas == null ? 0 : tableSchemas.size(),
+ stored != null && stored.getTable() != null);
// changeType is not consumed inside cdc_client — downstream only
reads getTable() and
// serializeTableSchemas does not persist it — so ALTER is used
uniformly, including for the
Review Comment:
This advances the PostgreSQL schema baseline even though no Doris DDL is
emitted. A non-ADD/DROP Relation change reaches this branch, and the rename
guard below has the same pattern for simultaneous ADD+DROP: `updatedSchemas`
gets the fresh source schema, the DDL list is empty, and `PipelineCoordinator`
still calls `applySchemaChange()`. After that, FE believes the target has the
new source schema while the Doris table was left unchanged. Please keep
skipped/manual PostgreSQL changes from being persisted as applied, or require
the corresponding Doris schema change before committing offsets against the
fresh baseline.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/PostgresDebeziumJsonDeserializer.java:
##########
@@ -200,7 +209,7 @@ private DeserializeResult handleSchemaChangeEvent(
targetTable,
col.name(),
SchemaChangeHelper.buildAddColumnSql(
Review Comment:
This path acknowledges that existing Doris rows are not backfilled, but it
still treats PostgreSQL `ADD COLUMN ... DEFAULT` / `NOT NULL` as an applied
schema change and advances the baseline. PostgreSQL returns the default for
existing source rows after such an ADD, while Doris keeps those rows `NULL` and
omits the source default/constraint. If backfilling is not supported here,
these DEFAULT/NOT NULL ADDs should be treated as unsupported/manual instead of
being persisted as fully applied; otherwise the Doris DDL needs to carry
equivalent default/backfill/nullability behavior.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/SchemaChangeHelper.java:
##########
@@ -164,22 +261,15 @@ static String pgTypeNameToDorisType(String pgTypeName,
int length, int scale) {
return DorisType.DOUBLE;
case "numeric":
{
- int p = length > 0 ? Math.min(length, 38) : 38;
+ if (length > MAX_DECIMAL128_PRECISION) {
+ return DorisType.STRING;
+ }
+ int p = length > 0 ? length : 38;
Review Comment:
This default for PostgreSQL `numeric` does not match initial table creation.
For an unbounded `numeric`, the JDBC path passes precision 0 into
`createDecimalOrStringType()`, which falls back to Doris `STRING`; this
schema-change mapper instead turns missing precision/scale into `DECIMAL(38,
9)`. The same upstream column can therefore be unconstrained when present at
job creation but fixed to decimal(38,9) when added later. Please align this
with `JdbcPostgreSQLClient` and return `STRING` when the source precision is
not positive, including array coverage if `_numeric` remains supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
