Revert "[Fix](partial update) Fix core when successfully schema chang…

…e and load during a partial update (apache#26210)"

This reverts commit 6983736.

fe/fe-core/src/main/java/org/apache/doris/analysis/NativeInsertStmt.java

@@ -1032,9 +1032,6 @@ public void complete() throws UserException {
                     throw new DdlException("txn does not exist: " + transactionId);
                 }
                 txnState.addTableIndexes((OlapTable) targetTable);
-                if (isPartialUpdate) {
-                    txnState.setSchemaForPartialUpdate((OlapTable) targetTable);
-                }
             }
         }
     }

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java

@@ -235,9 +235,6 @@ brokerFileGroups, getDeadlineMs(), getExecMemLimit(),
                     throw new UserException("txn does not exist: " + transactionId);
                 }
                 txnState.addTableIndexes(table);
-                if (isPartialUpdate()) {
-                    txnState.setSchemaForPartialUpdate(table);
-                }
             }
         } finally {
             MetaLockUtils.readUnlockTables(tableList);

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -896,9 +896,6 @@ public TExecPlanFragmentParams plan(TUniqueId loadId, long txnId) throws UserExc
                 throw new MetaNotFoundException("txn does not exist: " + txnId);
             }
             txnState.addTableIndexes(planner.getDestTable());
-            if (isPartialUpdate) {
-                txnState.setSchemaForPartialUpdate((OlapTable) table);
-            }
 
             return planParams;
         } finally {
@@ -919,9 +916,6 @@ public TPipelineFragmentParams planForPipeline(TUniqueId loadId, long txnId) thr
                 throw new MetaNotFoundException("txn does not exist: " + txnId);
             }
             txnState.addTableIndexes(planner.getDestTable());
-            if (isPartialUpdate) {
-                txnState.setSchemaForPartialUpdate((OlapTable) table);
-            }
 
             return planParams;
         } finally {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java

@@ -182,9 +182,6 @@ public void finalizeSink(DataSink sink, boolean isPartialUpdate, boolean isFromI
             throw new AnalysisException("txn does not exist: " + txnId);
         }
         state.addTableIndexes((OlapTable) table);
-        if (isPartialUpdate) {
-            state.setSchemaForPartialUpdate((OlapTable) table);
-        }
     }
 
     /**

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -2049,9 +2049,6 @@ private TExecPlanFragmentParams generatePlanFragmentParams(TStreamLoadPutRequest
                     throw new UserException("txn does not exist: " + request.getTxnId());
                 }
                 txnState.addTableIndexes(table);
-                if (request.isPartialUpdate()) {
-                    txnState.setSchemaForPartialUpdate(table);
-                }
             }
             plan.setTableName(table.getName());
             plan.query_options.setFeProcessUuid(ExecuteEnv.getInstance().getProcessUUID());
@@ -2115,9 +2112,6 @@ private TPipelineFragmentParams generatePipelineStreamLoadPut(TStreamLoadPutRequ
                     throw new UserException("txn does not exist: " + request.getTxnId());
                 }
                 txnState.addTableIndexes(table);
-                if (request.isPartialUpdate()) {
-                    txnState.setSchemaForPartialUpdate(table);
-                }
             }
             return plan;
         } finally {

fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java

@@ -682,35 +682,6 @@ public void commitTransaction(List<Table> tableList, long transactionId, List<Ta
                     + "] is prepare, not pre-committed.");
         }
 
-        if (transactionState.isPartialUpdate()) {
-            if (is2PC) {
-                Iterator<TableCommitInfo> tableCommitInfoIterator
-                        = transactionState.getIdToTableCommitInfos().values().iterator();
-                while (tableCommitInfoIterator.hasNext()) {
-                    TableCommitInfo tableCommitInfo = tableCommitInfoIterator.next();
-                    long tableId = tableCommitInfo.getTableId();
-                    OlapTable table = (OlapTable) db.getTableNullable(tableId);
-                    if (table != null && table instanceof OlapTable) {
-                        if (!transactionState.checkSchemaCompatibility((OlapTable) table)) {
-                            throw new TransactionCommitFailedException("transaction [" + transactionId
-                                    + "] check schema compatibility failed, partial update can't commit with"
-                                            + " old schema sucessfully .");
-                        }
-                    }
-                }
-            } else {
-                for (Table table : tableList) {
-                    if (table instanceof OlapTable) {
-                        if (!transactionState.checkSchemaCompatibility((OlapTable) table)) {
-                            throw new TransactionCommitFailedException("transaction [" + transactionId
-                                    + "] check schema compatibility failed, partial update can't commit with"
-                                            + " old schema sucessfully .");
-                        }
-                    }
-                }
-            }
-        }
-
         Set<Long> errorReplicaIds = Sets.newHashSet();
         Set<Long> totalInvolvedBackends = Sets.newHashSet();
         Map<Long, Set<Long>> tableToPartition = new HashMap<>();

fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java

@@ -17,9 +17,7 @@
 
 package org.apache.doris.transaction;
 
-import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Env;
-import org.apache.doris.catalog.MaterializedIndexMeta;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.FeMetaVersion;
@@ -47,10 +45,8 @@
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Comparator;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -276,24 +272,6 @@ public String toString() {
     // no need to persist.
     private String errMsg = "";
 
-    public class SchemaInfo {
-        public List<Column> schema;
-        public int schemaVersion;
-
-        public SchemaInfo(OlapTable olapTable) {
-            Map<Long, MaterializedIndexMeta> indexIdToMeta = olapTable.getIndexIdToMeta();
-            for (MaterializedIndexMeta indexMeta : indexIdToMeta.values()) {
-                schema = indexMeta.getSchema();
-                schemaVersion = indexMeta.getSchemaVersion();
-                break;
-            }
-        }
-    }
-
-    private boolean isPartialUpdate = false;
-    // table id -> schema info
-    private Map<Long, SchemaInfo> txnSchemas = new HashMap<>();
-
     public TransactionState() {
         this.dbId = -1;
         this.tableIdList = Lists.newArrayList();
@@ -773,46 +751,4 @@ public void pruneAfterVisible() {
         tableIdToTotalNumDeltaRows.clear();
     }
 
-    public void setSchemaForPartialUpdate(OlapTable olapTable) {
-        // the caller should hold the read lock of the table
-        isPartialUpdate = true;
-        txnSchemas.put(olapTable.getId(), new SchemaInfo(olapTable));
-    }
-
-    public boolean isPartialUpdate() {
-        return isPartialUpdate;
-    }
-
-    public SchemaInfo getTxnSchema(long id) {
-        return txnSchemas.get(id);
-    }
-
-    public boolean checkSchemaCompatibility(OlapTable olapTable) {
-        SchemaInfo currentSchemaInfo = new SchemaInfo(olapTable);
-        SchemaInfo txnSchemaInfo = txnSchemas.get(olapTable.getId());
-        if (txnSchemaInfo == null) {
-            return true;
-        }
-        if (txnSchemaInfo.schemaVersion >= currentSchemaInfo.schemaVersion) {
-            return true;
-        }
-        for (Column txnCol : txnSchemaInfo.schema) {
-            if (!txnCol.isVisible() || !txnCol.getType().isStringType()) {
-                continue;
-            }
-            int uniqueId = txnCol.getUniqueId();
-            Optional<Column> currentCol = currentSchemaInfo.schema.stream()
-                    .filter(col -> col.getUniqueId() == uniqueId).findFirst();
-            // for now Doris's light schema change only supports adding columns,
-            // dropping columns, and type conversions that increase the varchar length
-            if (currentCol.isPresent() && currentCol.get().getType().isStringType()) {
-                if (currentCol.get().getStrLen() != txnCol.getStrLen()) {
-                    LOG.warn("Check schema compatibility failed, txnId={}, table={}",
-                            transactionId, olapTable.getName());
-                    return false;
-                }
-            }
-        }
-        return true;
-    }
 }