Repository: NetEase/arctic (java)

Issue:
[Improvement]: Fix MySQL timestamp type missing default value
### Search before asking

- [X] I have searched in the [issues](https://github.com/NetEase/arctic/issues?q=is%3Aissue) and found no similar issues.


### What would you like to be improved?

In MySQL, if a TIMESTAMP field is defined in a table without specifying a default value, it will automatically be set to '0000-00-00 00:00:00' when inserting a row into that field. However, if strict sql mode is enabled, MySQL will prevent the insertion of '0000-00-00 00:00:00' because it is an invalid datetime value. In strict mode, only valid datetime values can be inserted into TIMESTAMP fields, otherwise an error will be thrown.

### How should we improve?

To avoid using a TIMESTAMP without a default value.

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Subtasks

_No response_

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)


Pull request description:
[ARCTIC-1634] avoid using timestamp without default value for MySQL
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
-->

fix #1634 

## Brief change log

  - replace timestamp without default value with datetime default null

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? (not documented)


Patch:
diff --git a/ams/server/src/main/resources/mysql/ams-mysql-init.sql b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
index 978592e40..94fc026ab 100644
--- a/ams/server/src/main/resources/mysql/ams-mysql-init.sql
+++ b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
@@ -102,9 +102,9 @@ CREATE TABLE `table_runtime`
     `current_change_snapshotId`     bigint(20) DEFAULT NULL COMMENT 'Change table current snapshot id',
     `last_optimized_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized snapshot id',
     `last_optimized_change_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized change snapshot id',
-    `last_major_optimizing_time`    timestamp COMMENT 'Latest Major Optimize time for all partitions',
-    `last_minor_optimizing_time`    timestamp COMMENT 'Latest Minor Optimize time for all partitions',
-    `last_full_optimizing_time`     timestamp COMMENT 'Latest Full Optimize time for all partitions',
+    `last_major_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Major Optimize time for all partitions',
+    `last_minor_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Minor Optimize time for all partitions',
+    `last_full_optimizing_time`     timestamp NULL DEFAULT NULL COMMENT 'Latest Full Optimize time for all partitions',
     `optimizing_status`             varchar(20) DEFAULT 'Idle' COMMENT 'Table optimize status: MajorOptimizing, MinorOptimizing, Pending, Idle',
     `optimizing_status_start_time`  timestamp default CURRENT_TIMESTAMP COMMENT 'Table optimize status start time',
     `optimizing_process_id`         bigint(20) NOT NULL COMMENT 'optimizing_procedure UUID',
@@ -126,8 +126,8 @@ CREATE TABLE `table_optimizing_process`
     `target_change_snapshot_id`     bigint(20) NOT NULL,
     `status`                        varchar(10) NOT NULL COMMENT 'Direct to TableOptimizingStatus',
     `optimizing_type`               varchar(10) NOT NULL COMMENT 'Optimize type: Major, Minor',
-    `plan_time`                     timestamp default CURRENT_TIMESTAMP COMMENT 'First plan time',
-    `end_time`                      datetime(3) DEFAULT NULL COMMENT 'finish time or failed time',
+    `plan_time`                     timestamp DEFAULT CURRENT_TIMESTAMP COMMENT 'First plan time',
+    `end_time`                      timestamp NULL DEFAULT NULL COMMENT 'finish time or failed time',
     `fail_reason`                   varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
     `rewrite_input`                 longblob DEFAULT NULL COMMENT 'rewrite files input',
     `summary`                       mediumtext COMMENT 'Max change transaction id of these tasks',
@@ -144,9 +144,9 @@ CREATE TABLE `task_runtime`
     `retry_num`                 int(11) DEFAULT NULL COMMENT 'Retry times',
     `table_id`                  bigint(20) NOT NULL,
     `partition_data`            varchar(128)  DEFAULT NULL COMMENT 'Partition data',
-    `create_time`               datetime(3) DEFAULT NULL COMMENT 'Task create time',
-    `start_time`                datetime(3) DEFAULT NULL COMMENT 'Time when task start waiting to execute',
-    `end_time`                  datetime(3) DEFAULT NULL COMMENT 'Time when task finished',
+    `create_time`               timestamp NULL DEFAULT NULL COMMENT 'Task create time',
+    `start_time`                timestamp NULL DEFAULT NULL COMMENT 'Time when task start waiting to execute',
+    `end_time`                  timestamp NULL DEFAULT NULL COMMENT 'Time when task finished',
     `cost_time`                 bigint(20) DEFAULT NULL,
     `status`                    varchar(16)   DEFAULT NULL  COMMENT 'Optimize Status: Init, Pending, Executing, Failed, Prepared, Committed',
     `fail_reason`               varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
@@ -178,7 +178,7 @@ CREATE TABLE `api_tokens`
     `id`         int(11) NOT NULL AUTO_INCREMENT,
     `apikey`     varchar(256) NOT NULL COMMENT 'openapi client public key',
     `secret`     varchar(256) NOT NULL COMMENT 'The key used by the client to generate the request signature',
-    `apply_time` datetime DEFAULT NULL COMMENT 'apply time',
+    `apply_time` timestamp NULL DEFAULT NULL COMMENT 'apply time',
     PRIMARY KEY (`id`) USING BTREE,
     UNIQUE KEY `account_unique` (`apikey`) USING BTREE COMMENT 'account unique'
 ) ENGINE=InnoDB AUTO_INCREMENT=33 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='Openapi  secret';
@@ -198,8 +198,8 @@ CREATE TABLE `table_blocker` (
   `db_name` varchar(128) NOT NULL COMMENT 'Database name',
   `table_name` varchar(128) NOT NULL COMMENT 'Table name',
   `operations` varchar(128) NOT NULL COMMENT 'Blocked operations',
-  `create_time` datetime(3) DEFAULT NULL COMMENT 'Blocker create time',
-  `expiration_time` datetime(3) DEFAULT NULL COMMENT 'Blocker expiration time',
+  `create_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker create time',
+  `expiration_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker expiration time',
   `properties` mediumtext COMMENT 'Blocker properties',
   PRIMARY KEY (`blocker_id`),
   KEY `table_index` (`catalog_name`,`db_name`,`table_name`)
diff --git a/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql b/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
index 337acb83f..1e3a67a88 100644
--- a/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
+++ b/ams/server/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql
@@ -94,9 +94,9 @@ CREATE TABLE `table_runtime`
     `current_change_snapshotId`     bigint(20) DEFAULT NULL COMMENT 'Change table current snapshot id',
     `last_optimized_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized snapshot id',
     `last_optimized_change_snapshotId`     bigint(20) NOT NULL DEFAULT '-1' COMMENT 'last optimized change snapshot id',
-    `last_major_optimizing_time`    timestamp COMMENT 'Latest Major Optimize time for all partitions',
-    `last_minor_optimizing_time`    timestamp COMMENT 'Latest Minor Optimize time for all partitions',
-    `last_full_optimizing_time`     timestamp COMMENT 'Latest Full Optimize time for all partitions',
+    `last_major_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Major Optimize time for all partitions',
+    `last_minor_optimizing_time`    timestamp NULL DEFAULT NULL COMMENT 'Latest Minor Optimize time for all partitions',
+    `last_full_optimizing_time`     timestamp NULL DEFAULT NULL COMMENT 'Latest Full Optimize time for all partitions',
     `optimizing_status`             varchar(20) DEFAULT 'Idle' COMMENT 'Table optimize status: MajorOptimizing, MinorOptimizing, Pending, Idle',
     `optimizing_status_start_time`  timestamp default CURRENT_TIMESTAMP COMMENT 'Table optimize status start time',
     `optimizing_process_id`         bigint(20) NOT NULL COMMENT 'optimizing_procedure UUID',
@@ -116,9 +116,9 @@ CREATE TABLE `task_runtime`
     `retry_num`                 int(11) DEFAULT NULL COMMENT 'Retry times',
     `table_id`                  bigint(20) NOT NULL,
     `partition_data`            varchar(128)  DEFAULT NULL COMMENT 'Partition data',
-    `create_time`               datetime(3) DEFAULT NULL COMMENT 'Task create time',
-    `start_time`                datetime(3) DEFAULT NULL COMMENT 'Time when task start waiting to execute',
-    `end_time`                  datetime(3) DEFAULT NULL COMMENT 'Time when task finished',
+    `create_time`               timestamp NULL DEFAULT NULL COMMENT 'Task create time',
+    `start_time`                timestamp NULL DEFAULT NULL COMMENT 'Time when task start waiting to execute',
+    `end_time`                  timestamp NULL DEFAULT NULL COMMENT 'Time when task finished',
     `cost_time`                 bigint(20) DEFAULT NULL,
     `status`                    varchar(16)   DEFAULT NULL  COMMENT 'Optimize Status: Init, Pending, Executing, Failed, Prepared, Committed',
     `fail_reason`               varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
@@ -143,8 +143,8 @@ CREATE TABLE `table_optimizing_process`
     `target_change_snapshot_id`     bigint(20) NOT NULL,
     `status`                        varchar(10) NOT NULL COMMENT 'Direct to TableOptimizingStatus',
     `optimizing_type`               varchar(10) NOT NULL COMMENT 'Optimize type: Major, Minor',
-    `plan_time`                     timestamp default CURRENT_TIMESTAMP COMMENT 'First plan time',
-    `end_time`                      datetime(3) DEFAULT NULL COMMENT 'finish time or failed time',
+    `plan_time`                     timestamp DEFAULT CURRENT_TIMESTAMP COMMENT 'First plan time',
+    `end_time`                      timestamp NULL DEFAULT NULL COMMENT 'finish time or failed time',
     `fail_reason`                   varchar(4096) DEFAULT NULL COMMENT 'Error message after task failed',
     `rewrite_input`                 longblob DEFAULT NULL COMMENT 'rewrite files input',
     `summary`                       mediumtext COMMENT 'Max change transaction id of these tasks',
@@ -168,6 +168,11 @@ CREATE TABLE `optimizing_task_quota`
     KEY  `table_index` (`table_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Optimize task basic information';
 
+-- modify to timestamp
+ALTER TABLE `table_blocker` MODIFY `create_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker create time';
+ALTER TABLE `table_blocker` MODIFY `expiration_time` timestamp NULL DEFAULT NULL COMMENT 'Blocker expiration time';
+ALTER TABLE `api_tokens` MODIFY `apply_time` timestamp NULL DEFAULT NULL COMMENT 'apply time';
+
 -- init table_runtime
 INSERT INTO table_runtime (table_id, catalog_name, db_name, table_name, current_snapshot_id, current_change_snapshotId,
 last_optimized_snapshotId, last_optimized_change_snapshotId,
diff --git a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
index c2d42f443..826f03e79 100644
--- a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
+++ b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
@@ -114,6 +114,10 @@ public class ArcticDataFiles {
               field.name().equals(parts[0]),
           "Invalid partition: %s", partitions[i]);
 
+      if ("null".equals(parts[1])) {
+        parts[1] = null;
+      }
+
       data.set(i, ArcticDataFiles.fromPartitionString(field, spec.partitionType().fieldType(parts[0]), parts[1]));
     }
 


Test patch:
diff --git a/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java b/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
index 70159bd29..c9c21c997 100644
--- a/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
+++ b/ams/server/src/test/java/com/netease/arctic/server/optimizing/flow/view/KeyedTableDataView.java
@@ -248,7 +248,11 @@ public class KeyedTableDataView extends AbstractTableDataView {
       Object o1 = r1.get(i);
       Object o2 = r2.get(i);
       boolean equals;
-      if (o1 instanceof OffsetDateTime) {
+      if (o1 == null && o2 == null) {
+        return true;
+      } else if (o1 == null || o2 == null) {
+        return false;
+      } else if (o1 instanceof OffsetDateTime) {
         equals = ((OffsetDateTime) o1).isEqual((OffsetDateTime) o2);
       } else {
         equals = o1.equals(o2);
diff --git a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
index c9a592770..4f83a47f0 100644
--- a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
+++ b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
@@ -128,4 +128,16 @@ public class TestArcticDataFiles {
     p2.set(partitionData);
     Assert.assertEquals(p1, p2);
   }
+
+  @Test
+  public void testNullPartition() {
+    Schema schema = new Schema(
+        Types.NestedField.required(1, "name", Types.StringType.get())
+    );
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("name").build();
+    PartitionKey partitionKey = new PartitionKey(spec, schema);
+    String partitionPath = "name=null";
+    StructLike partitionData = ArcticDataFiles.data(spec, partitionPath);
+    Assert.assertNull(partitionData.get(0, Types.StringType.get().typeId().javaClass()));
+  }
 }