Skip to content

The microseconds default value of '0000-00-00 00:00:00.000000' for MySQL TIMESTAMP fields is not supported in downstream systems. #4251

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lvyanquan merged 5 commits into from
Mar 30, 2026
Merged

The microseconds default value of '0000-00-00 00:00:00.000000' for MySQL TIMESTAMP fields is not supported in downstream systems. #4251

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d2ca03c
add Fix microseconds default value
yunhaoww Jan 31, 2026
c2a0b75
add tests for Paimon and StarRocks
yunhaoww Jan 31, 2026
7c6b4b6
lint code
yunhaoww Jan 31, 2026
63d4375
lint
yunhaoww Jan 31, 2026
87e2041
Merge branch 'master' into fixMicroSecondsDefaultValue
yunhaoww Mar 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisMetadataApplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ private String convertInvalidTimestampDefaultValue(String defaultValue, DataType
|| dataType instanceof TimestampType
|| dataType instanceof ZonedTimestampType) {

if (DorisSchemaUtils.INVALID_OR_MISSING_DATATIME.equals(defaultValue)) {
if (defaultValue.startsWith(DorisSchemaUtils.INVALID_OR_MISSING_DATATIME)) {
return DorisSchemaUtils.DEFAULT_DATETIME;
}
}
Expand Down
102 changes: 102 additions & 0 deletions .../src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisMetadataApplierITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -630,6 +630,108 @@ void testMysqlDefaultTimestampValueConversionInAddColumn(boolean batchMode) thro
assertEqualsInOrder(expected, actual);
}

/** Microsecond variant: '0000-00-00 00:00:00.000000'. */
private static final String INVALID_DATETIME_WITH_MICROS = "0000-00-00 00:00:00.000000";

@ParameterizedTest(name = "batchMode: {0}")
@ValueSource(booleans = {true, false})
void testMysqlDefaultTimestampValueWithMicrosInCreateTable(boolean batchMode) throws Exception {
TableId tableId =
TableId.tableId(
DorisContainer.DORIS_DATABASE_NAME, DorisContainer.DORIS_TABLE_NAME);

Schema schema =
Schema.newBuilder()
.column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
.column(new PhysicalColumn("name", DataTypes.VARCHAR(50), null))
.column(
new PhysicalColumn(
"created_time",
DataTypes.TIMESTAMP(6),
null,
INVALID_DATETIME_WITH_MICROS))
.column(
new PhysicalColumn(
"updated_time",
DataTypes.TIMESTAMP_LTZ(6),
null,
INVALID_DATETIME_WITH_MICROS))
.primaryKey("id")
.build();

runJobWithEvents(
Collections.singletonList(new CreateTableEvent(tableId, schema)), batchMode);

List<String> actual = inspectTableSchema(tableId);

List<String> expected =
Arrays.asList(
"id | INT | Yes | true | null",
"name | VARCHAR(150) | Yes | false | null",
"created_time | DATETIME(6) | Yes | false | "
+ DorisSchemaUtils.DEFAULT_DATETIME,
"updated_time | DATETIME(6) | Yes | false | "
+ DorisSchemaUtils.DEFAULT_DATETIME);

assertEqualsInOrder(expected, actual);
}

@ParameterizedTest(name = "batchMode: {0}")
@ValueSource(booleans = {true, false})
void testMysqlDefaultTimestampValueWithMicrosInAddColumn(boolean batchMode) throws Exception {
TableId tableId =
TableId.tableId(
DorisContainer.DORIS_DATABASE_NAME, DorisContainer.DORIS_TABLE_NAME);

Schema initialSchema =
Schema.newBuilder()
.column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
.column(new PhysicalColumn("name", DataTypes.VARCHAR(50), null))
.primaryKey("id")
.build();

List<Event> events = new ArrayList<>();
events.add(new CreateTableEvent(tableId, initialSchema));

PhysicalColumn createdTimeCol =
new PhysicalColumn(
"created_time", DataTypes.TIMESTAMP(6), null, INVALID_DATETIME_WITH_MICROS);

PhysicalColumn updatedTimeCol =
new PhysicalColumn(
"updated_time",
DataTypes.TIMESTAMP_LTZ(6),
null,
INVALID_DATETIME_WITH_MICROS);

events.add(
new AddColumnEvent(
tableId,
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(createdTimeCol))));

events.add(
new AddColumnEvent(
tableId,
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(updatedTimeCol))));

runJobWithEvents(events, batchMode);

List<String> actual = inspectTableSchema(tableId);

List<String> expected =
Arrays.asList(
"id | INT | Yes | true | null",
"name | VARCHAR(150) | Yes | false | null",
"created_time | DATETIME(6) | Yes | false | "
+ DorisSchemaUtils.DEFAULT_DATETIME,
"updated_time | DATETIME(6) | Yes | false | "
+ DorisSchemaUtils.DEFAULT_DATETIME);

assertEqualsInOrder(expected, actual);
}

private void runJobWithEvents(List<Event> events, boolean batchMode) throws Exception {
DataStream<Event> stream = env.fromData(events, new EventTypeInfo()).setParallelism(1);

Expand Down
2 changes: 1 addition & 1 deletion ...aimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/SchemaChangeProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ private static String convertInvalidTimestampDefaultValue(
|| dataType instanceof TimestampType
|| dataType instanceof ZonedTimestampType) {

if (INVALID_OR_MISSING_DATATIME.equals(defaultValue)) {
if (defaultValue.startsWith(INVALID_OR_MISSING_DATATIME)) {
return DEFAULT_DATETIME;
}
}
Expand Down
54 changes: 54 additions & 0 deletions .../src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -669,4 +669,58 @@ public void testMysqlDefaultTimestampValueConversionInAddColumn()

Assertions.assertThat(table).isNotNull();
}

/** Microsecond variant: '0000-00-00 00:00:00.000000'. */
private static final String INVALID_DATETIME_WITH_MICROS = "0000-00-00 00:00:00.000000";

@Test
public void testMysqlDefaultTimestampValueWithMicrosInAddColumn()
throws SchemaEvolveException,
Catalog.TableNotExistException,
Catalog.DatabaseNotEmptyException,
Catalog.DatabaseNotExistException {
initialize("filesystem");
Map<String, String> tableOptions = new HashMap<>();
tableOptions.put("bucket", "-1");
MetadataApplier metadataApplier =
new PaimonMetadataApplier(catalogOptions, tableOptions, new HashMap<>());

CreateTableEvent createTableEvent =
new CreateTableEvent(
TableId.parse("test.timestamp_micros_test"),
org.apache.flink.cdc.common.schema.Schema.newBuilder()
.physicalColumn(
"id",
org.apache.flink.cdc.common.types.DataTypes.INT().notNull())
.physicalColumn(
"name",
org.apache.flink.cdc.common.types.DataTypes.STRING())
.primaryKey("id")
.build());
metadataApplier.applySchemaChange(createTableEvent);

List<AddColumnEvent.ColumnWithPosition> addedColumns = new ArrayList<>();
addedColumns.add(
AddColumnEvent.last(
Column.physicalColumn(
"created_time",
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP(6),
null,
INVALID_DATETIME_WITH_MICROS)));
addedColumns.add(
AddColumnEvent.last(
Column.physicalColumn(
"updated_time",
org.apache.flink.cdc.common.types.DataTypes.TIMESTAMP_LTZ(6),
null,
INVALID_DATETIME_WITH_MICROS)));

AddColumnEvent addColumnEvent =
new AddColumnEvent(TableId.parse("test.timestamp_micros_test"), addedColumns);
metadataApplier.applySchemaChange(addColumnEvent);

Table table = catalog.getTable(Identifier.fromString("test.timestamp_micros_test"));

Assertions.assertThat(table).isNotNull();
}
}
101 changes: 101 additions & 0 deletions ...t/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -582,4 +582,105 @@ void testMysqlDefaultTimestampValueConversionInAddColumn() throws Exception {

assertEqualsInOrder(expected, actual);
}

/** Microsecond variant: '0000-00-00 00:00:00.000000'. */
private static final String INVALID_DATETIME_WITH_MICROS = "0000-00-00 00:00:00.000000";

@Test
void testMysqlDefaultTimestampValueWithMicrosInCreateTable() throws Exception {
TableId tableId =
TableId.tableId(
StarRocksContainer.STARROCKS_DATABASE_NAME,
StarRocksContainer.STARROCKS_TABLE_NAME);

Schema schema =
Schema.newBuilder()
.column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
.column(new PhysicalColumn("name", DataTypes.VARCHAR(50), null))
.column(
new PhysicalColumn(
"created_time",
DataTypes.TIMESTAMP(6),
null,
INVALID_DATETIME_WITH_MICROS))
.column(
new PhysicalColumn(
"updated_time",
DataTypes.TIMESTAMP_LTZ(6),
null,
INVALID_DATETIME_WITH_MICROS))
.primaryKey("id")
.build();

runJobWithEvents(Collections.singletonList(new CreateTableEvent(tableId, schema)));

List<String> actual = inspectTableSchema(tableId);

List<String> expected =
Arrays.asList(
"id | int | NO | true | null",
"name | varchar(150) | YES | false | null",
"created_time | datetime | YES | false | "
+ StarRocksUtils.DEFAULT_DATETIME,
"updated_time | datetime | YES | false | "
+ StarRocksUtils.DEFAULT_DATETIME);

assertEqualsInOrder(expected, actual);
}

@Test
void testMysqlDefaultTimestampValueWithMicrosInAddColumn() throws Exception {
TableId tableId =
TableId.tableId(
StarRocksContainer.STARROCKS_DATABASE_NAME,
StarRocksContainer.STARROCKS_TABLE_NAME);

Schema initialSchema =
Schema.newBuilder()
.column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
.column(new PhysicalColumn("name", DataTypes.VARCHAR(50), null))
.primaryKey("id")
.build();

List<Event> events = new ArrayList<>();
events.add(new CreateTableEvent(tableId, initialSchema));

PhysicalColumn createdTimeCol =
new PhysicalColumn(
"created_time", DataTypes.TIMESTAMP(6), null, INVALID_DATETIME_WITH_MICROS);

PhysicalColumn updatedTimeCol =
new PhysicalColumn(
"updated_time",
DataTypes.TIMESTAMP_LTZ(6),
null,
INVALID_DATETIME_WITH_MICROS);

events.add(
new AddColumnEvent(
tableId,
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(createdTimeCol))));

events.add(
new AddColumnEvent(
tableId,
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(updatedTimeCol))));

runJobWithEvents(events);

List<String> actual = inspectTableSchema(tableId);

List<String> expected =
Arrays.asList(
"id | int | NO | true | null",
"name | varchar(150) | YES | false | null",
"created_time | datetime | YES | false | "
+ StarRocksUtils.DEFAULT_DATETIME,
"updated_time | datetime | YES | false | "
+ StarRocksUtils.DEFAULT_DATETIME);

assertEqualsInOrder(expected, actual);
}
}
Loading