Repository: StarRocks/starrocks
Issue: Wrong result return while date type of column is `datetime`.

<!-- (At least include the following, feel free to add if you have more content) -->
insert and delete error.
1. Precision problem of `datetime` when insert into table values.
2. Delete error('ERROR 1064 (HY000): Invalid column value[null] for column k2')
### Steps to reproduce the behavior (Required)
```
mysql> CREATE TABLE test_string2 (`k1` string not null,k2 datetime) ENGINE=OLAP duplicate key(k1) DISTRIBUTED BY HASH(`k1`) BUCKETS 3 PROPERTIES ("replication_num" = "1");
Query OK, 0 rows affected (0.01 sec)

mysql> insert into test_string2 values('2021-12-29 16:58:56.032000','2021-12-29 16:58:56.032000');
Query OK, 1 row affected (0.03 sec)
{'label':'insert_5e9563d2-689f-11ec-8fab-00163e10c9d6', 'status':'VISIBLE', 'txnId':'33055'}

mysql> select * from test_string2;
+----------------------------+----------------------------+
| k1                         | k2                         |
+----------------------------+----------------------------+
| 2021-12-29 16:58:56.032000 | 2021-12-29 16:58:56.032000 |
+----------------------------+----------------------------+
1 row in set (0.01 sec)

mysql> delete from test_string2 where k2='2021-12-29 16:58:56.032000';
ERROR 1064 (HY000): Invalid column value[null] for column k2
mysql> delete from test_string2 where k1='2021-12-29 16:58:56.032000';
Query OK, 0 rows affected (0.02 sec)
{'label':'delete_a088fb14-0748-44ce-b354-18e3b66fc20d', 'status':'VISIBLE', 'txnId':'33056'}

mysql> select * from test_string2;
Empty set (0.00 sec)
```

-mysql
```
mysql>  CREATE TABLE test_string2 (`k1` datetime not null,k2 datetime);
Query OK, 0 rows affected (0.02 sec)

mysql> insert into test_string2 values('2021-12-29 16:58:56.032000','2021-12-29 16:58:56.032000');
Query OK, 1 row affected (0.00 sec)

mysql> select * from test_string2;
+---------------------+---------------------+
| k1                  | k2                  |
+---------------------+---------------------+
| 2021-12-29 16:58:56 | 2021-12-29 16:58:56 |
+---------------------+---------------------+
1 row in set (0.00 sec)
```

### Expected behavior (Required)
```
mysql> select * from test_string2;
+---------------------+---------------------+
| k1                  | k2                  |
+---------------------+---------------------+
| 2021-12-29 16:58:56 | 2021-12-29 16:58:56 |
+---------------------+---------------------+
1 row in set (0.00 sec)
```
### Real behavior (Required)
```
mysql> select * from test_string2;
+----------------------------+----------------------------+
| k1                         | k2                         |
+----------------------------+----------------------------+
| 2021-12-29 16:58:56.032000 | 2021-12-29 16:58:56.032000 |
+----------------------------+----------------------------+
1 row in set (0.01 sec)
```
### StarRocks version (Required)
 - You can get the StarRocks version by executing SQL `select current_version()`
 - 2.1

Changed files:
fe/fe-core/src/main/java/com/starrocks/load/OlapDeleteJob.java, fe/fe-core/src/main/java/com/starrocks/task/PushTask.java, fe/fe-core/src/main/java/com/starrocks/common/MarkedCountDownLatch.java, fe/fe-core/src/main/java/com/starrocks/leader/LeaderImpl.java

diff --git a/fe/fe-core/src/main/java/com/starrocks/common/MarkedCountDownLatch.java b/fe/fe-core/src/main/java/com/starrocks/common/MarkedCountDownLatch.java
index 42f8a622dd..406ea029fc 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/MarkedCountDownLatch.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/MarkedCountDownLatch.java
@@ -64,6 +64,13 @@ public class MarkedCountDownLatch<K, V> extends CountDownLatch {
         return false;
     }
 
+    public synchronized boolean markedCountDown(K key, V value, Status status) {
+        if (st.ok()) {
+            st = status;
+        }
+        return markedCountDown(key, value);
+    }
+
     public synchronized List<Entry<K, V>> getLeftMarks() {
         return Lists.newArrayList(marks.entries());
     }
diff --git a/fe/fe-core/src/main/java/com/starrocks/leader/LeaderImpl.java b/fe/fe-core/src/main/java/com/starrocks/leader/LeaderImpl.java
index 146cf68054..16921c61da 100644
--- a/fe/fe-core/src/main/java/com/starrocks/leader/LeaderImpl.java
+++ b/fe/fe-core/src/main/java/com/starrocks/leader/LeaderImpl.java
@@ -224,12 +224,26 @@ public class LeaderImpl {
                 String errMsg = "task type: " + taskType + ", status_code: " + taskStatus.getStatus_code().toString() +
                         ", backendId: " + backendId + ", signature: " + signature;
                 task.setErrorMsg(errMsg);
+                LOG.warn(errMsg);
                 // We start to let FE perceive the task's error msg
                 if (taskType != TTaskType.MAKE_SNAPSHOT && taskType != TTaskType.UPLOAD
                         && taskType != TTaskType.DOWNLOAD && taskType != TTaskType.MOVE
                         && taskType != TTaskType.CLONE && taskType != TTaskType.PUBLISH_VERSION
                         && taskType != TTaskType.CREATE && taskType != TTaskType.UPDATE_TABLET_META_INFO
                         && taskType != TTaskType.DROP_AUTO_INCREMENT_MAP) {
+                    if (taskType == TTaskType.REALTIME_PUSH) {
+                        PushTask pushTask = (PushTask) task;
+                        if (pushTask.getPushType() == TPushType.DELETE) {
+                            LOG.info("remove push replica. tabletId: {}, backendId: {}", task.getSignature(),
+                                     pushTask.getBackendId());
+
+                            String failMsg = "Backend: " + task.getBackendId() + "Tablet: " + pushTask.getTabletId() +
+                                             " error msg: " + taskStatus.getError_msgs().toString();
+                            pushTask.countDownLatch(pushTask.getBackendId(), pushTask.getTabletId(), failMsg);
+                            AgentTaskQueue.removeTask(pushTask.getBackendId(), TTaskType.REALTIME_PUSH, 
+                                                      task.getSignature());
+                        }
+                    }
                     return result;
                 }
             }
diff --git a/fe/fe-core/src/main/java/com/starrocks/load/OlapDeleteJob.java b/fe/fe-core/src/main/java/com/starrocks/load/OlapDeleteJob.java
index 4c8b8f32df..11c9cc3836 100644
--- a/fe/fe-core/src/main/java/com/starrocks/load/OlapDeleteJob.java
+++ b/fe/fe-core/src/main/java/com/starrocks/load/OlapDeleteJob.java
@@ -54,6 +54,7 @@ import com.starrocks.common.DdlException;
 import com.starrocks.common.FeConstants;
 import com.starrocks.common.MarkedCountDownLatch;
 import com.starrocks.common.MetaNotFoundException;
+import com.starrocks.common.Status;
 import com.starrocks.common.UserException;
 import com.starrocks.qe.QueryStateException;
 import com.starrocks.server.GlobalStateMgr;
@@ -183,6 +184,7 @@ public class OlapDeleteJob extends DeleteJob {
         } finally {
             db.readUnlock();
         }
+        LOG.info("countDownLatch count: {}", countDownLatch.getCount());
 
         long timeoutMs = getTimeoutMs();
         LOG.info("waiting delete Job finish, signature: {}, timeout: {}", getTransactionId(), timeoutMs);
@@ -208,59 +210,65 @@ public class OlapDeleteJob extends DeleteJob {
         } catch (InterruptedException e) {
             LOG.warn("InterruptedException: ", e);
         }
+        LOG.info("delete job finish, countDownLatch count: {}", countDownLatch.getCount());
+ 
+        String errMsg = "";
+        List<Map.Entry<Long, Long>> unfinishedMarks = countDownLatch.getLeftMarks();
+        Status st = countDownLatch.getStatus();
+        // only show at most 5 results
+        List<Map.Entry<Long, Long>> subList = unfinishedMarks.subList(0, Math.min(unfinishedMarks.size(), 5));
+        if (!subList.isEmpty()) {
+            errMsg = "unfinished replicas: " + Joiner.on(", ").join(subList);
+        } else if (!st.ok()) {
+            errMsg = st.toString();
+        }
+        LOG.warn(errMsg);
 
-        if (!ok) {
-            String errMsg = "";
-            List<Map.Entry<Long, Long>> unfinishedMarks = countDownLatch.getLeftMarks();
-            // only show at most 5 results
-            List<Map.Entry<Long, Long>> subList = unfinishedMarks.subList(0, Math.min(unfinishedMarks.size(), 5));
-            if (!subList.isEmpty()) {
-                errMsg = "unfinished replicas: " + Joiner.on(", ").join(subList);
-            }
-            LOG.warn(errMsg);
-
-            try {
-                checkAndUpdateQuorum();
-            } catch (MetaNotFoundException e) {
-                cancel(DeleteHandler.CancelType.METADATA_MISSING, e.getMessage());
-                throw new DdlException(e.getMessage(), e);
-            }
-            DeleteState state = getState();
-            switch (state) {
-                case DELETING:
-                    LOG.warn("delete job timeout: transactionId {}, timeout {}, {}", getTransactionId(), timeoutMs,
-                            errMsg);
+        try {
+            checkAndUpdateQuorum();
+        } catch (MetaNotFoundException e) {
+            cancel(DeleteHandler.CancelType.METADATA_MISSING, e.getMessage());
+            throw new DdlException(e.getMessage(), e);
+        }
+        DeleteState state = getState();
+        switch (state) {
+            case DELETING:
+                LOG.warn("delete job failed: transactionId {}, timeout {}, {}", getTransactionId(), timeoutMs,
+                        errMsg);
+                if (countDownLatch.getCount() > 0) {
                     cancel(DeleteHandler.CancelType.TIMEOUT, "delete job timeout");
-                    throw new DdlException("failed to execute delete. transaction id " + getTransactionId() +
-                            ", timeout(ms) " + timeoutMs + ", " + errMsg);
-                case QUORUM_FINISHED:
-                case FINISHED:
-                    try {
-                        long nowQuorumTimeMs = System.currentTimeMillis();
-                        long endQuorumTimeoutMs = nowQuorumTimeMs + timeoutMs / 2;
-                        // if job's state is quorum_finished then wait for a period of time and commit it.
-                        while (getState() == DeleteState.QUORUM_FINISHED && endQuorumTimeoutMs > nowQuorumTimeMs) {
-                            checkAndUpdateQuorum();
-                            Thread.sleep(1000);
-                            nowQuorumTimeMs = System.currentTimeMillis();
-                            LOG.debug("wait for quorum finished delete job: {}, txn_id: {}", getId(),
-                                    getTransactionId());
-                        }
-                    } catch (MetaNotFoundException e) {
-                        cancel(DeleteHandler.CancelType.METADATA_MISSING, e.getMessage());
-                        throw new DdlException(e.getMessage(), e);
-                    } catch (InterruptedException e) {
-                        cancel(DeleteHandler.CancelType.UNKNOWN, e.getMessage());
-                        throw new DdlException(e.getMessage(), e);
+                } else {
+                    cancel(DeleteHandler.CancelType.UNKNOWN, "delete job failed");
+                }
+                throw new DdlException("failed to execute delete. transaction id " + getTransactionId() +
+                        ", timeout(ms) " + timeoutMs + ", " + errMsg);
+            case QUORUM_FINISHED:
+            case FINISHED:
+                try {
+                    long nowQuorumTimeMs = System.currentTimeMillis();
+                    long endQuorumTimeoutMs = nowQuorumTimeMs + timeoutMs / 2;
+                    // if job's state is quorum_finished then wait for a period of time and commit it.
+                    while (getState() == DeleteState.QUORUM_FINISHED && endQuorumTimeoutMs > nowQuorumTimeMs
+                          && countDownLatch.getCount() > 0) {
+                        checkAndUpdateQuorum();
+                        Thread.sleep(1000);
+                        nowQuorumTimeMs = System.currentTimeMillis();
+                        LOG.debug("wait for quorum finished delete job: {}, txn_id: {}", getId(),
+                                getTransactionId());
                     }
-                    commit(db, timeoutMs);
-                    break;
-                default:
-                    throw new IllegalStateException("wrong delete job state: " + state.name());
-            }
-        } else {
-            commit(db, timeoutMs);
+                } catch (MetaNotFoundException e) {
+                    cancel(DeleteHandler.CancelType.METADATA_MISSING, e.getMessage());
+                    throw new DdlException(e.getMessage(), e);
+                } catch (InterruptedException e) {
+                    cancel(DeleteHandler.CancelType.UNKNOWN, e.getMessage());
+                    throw new DdlException(e.getMessage(), e);
+                }
+                commit(db, timeoutMs);
+                break;
+            default:
+                throw new IllegalStateException("wrong delete job state: " + state.name());
         }
+
     }
 
     /**
diff --git a/fe/fe-core/src/main/java/com/starrocks/task/PushTask.java b/fe/fe-core/src/main/java/com/starrocks/task/PushTask.java
index bc5b8884a5..12a862afb6 100644
--- a/fe/fe-core/src/main/java/com/starrocks/task/PushTask.java
+++ b/fe/fe-core/src/main/java/com/starrocks/task/PushTask.java
@@ -43,6 +43,7 @@ import com.starrocks.analysis.LiteralExpr;
 import com.starrocks.analysis.Predicate;
 import com.starrocks.analysis.SlotRef;
 import com.starrocks.common.MarkedCountDownLatch;
+import com.starrocks.common.Status;
 import com.starrocks.thrift.TBrokerScanRange;
 import com.starrocks.thrift.TCondition;
 import com.starrocks.thrift.TDescriptorTable;
@@ -50,6 +51,7 @@ import com.starrocks.thrift.TPriority;
 import com.starrocks.thrift.TPushReq;
 import com.starrocks.thrift.TPushType;
 import com.starrocks.thrift.TResourceInfo;
+import com.starrocks.thrift.TStatusCode;
 import com.starrocks.thrift.TTabletType;
 import com.starrocks.thrift.TTaskType;
 import org.apache.logging.log4j.LogManager;
@@ -219,6 +221,15 @@ public class PushTask extends AgentTask {
         }
     }
 
+    public void countDownLatch(long backendId, long tabletId, String errMsg) {
+        if (this.latch != null) {
+            if (latch.markedCountDown(backendId, tabletId, new Status(TStatusCode.INTERNAL_ERROR, errMsg))) {
+                LOG.info("pushTask current latch count: {}. backend: {}, tablet:{}",
+                        latch.getCount(), backendId, tabletId);
+            }
+        }
+    }
+
     public long getReplicaId() {
         return replicaId;
     }