[feature](schema change) add columnsDesc to TPushReq for ligtht sc.

Lchangliang · Jun 17, 2022 · d4a3d2d · d4a3d2d
1 parent 9ca0d89
commit d4a3d2d
Show file tree

Hide file tree

Showing 7 changed files with 26 additions and 28 deletions.
diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
@@ -123,8 +123,7 @@ Status DeltaWriter::init() {
     _build_current_tablet_schema(_req.ptable_schema_param, _tablet->tablet_schema());
 
     RETURN_NOT_OK(_tablet->create_rowset_writer(_req.txn_id, _req.load_id, PREPARED, OVERLAPPING,
-                                                &_tablet->tablet_schema(), &_rowset_writer));
-    _tablet_schema = &(_tablet->tablet_schema());
+                                                _tablet_schema.get(), &_rowset_writer));
     _schema.reset(new Schema(*_tablet_schema));
     _reset_mem_table();
 

diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
@@ -268,19 +268,6 @@ class Tablet : public BaseTablet {
         return _tablet_meta->all_beta();
     }
 
-    void find_alpha_rowsets(std::vector<RowsetSharedPtr>* rowsets) const;
-
-    Status create_rowset_writer(const Version& version, const RowsetStatePB& rowset_state,
-                                const SegmentsOverlapPB& overlap,
-                                const doris::TabletSchema* tablet_schema,
-                                std::unique_ptr<RowsetWriter>* rowset_writer);
-
-    Status create_rowset_writer(const int64_t& txn_id, const PUniqueId& load_id,
-                                const RowsetStatePB& rowset_state, const SegmentsOverlapPB& overlap,
-                                const doris::TabletSchema* tablet_schema,
-                                std::unique_ptr<RowsetWriter>* rowset_writer);
-
-    Status create_rowset(RowsetMetaSharedPtr rowset_meta, RowsetSharedPtr* rowset);
     const TabletSchema& tablet_schema() const override;
 
 private:

diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
@@ -65,6 +65,7 @@
 import org.apache.doris.task.AgentTaskExecutor;
 import org.apache.doris.task.AgentTaskQueue;
 import org.apache.doris.task.PushTask;
+import org.apache.doris.thrift.TColumn;
 import org.apache.doris.thrift.TPriority;
 import org.apache.doris.thrift.TPushType;
 import org.apache.doris.thrift.TTaskType;
@@ -216,6 +217,12 @@ public void process(DeleteStmt stmt) throws DdlException, QueryStateException {
                 // to make sure that the delete transaction can be done successfully.
                 txnState.addTableIndexes(olapTable);
 
+                //for light schema change
+                List<TColumn> columnsDesc = new ArrayList<TColumn>();
+                for (Column column : olapTable.getFullSchema()) {
+                    columnsDesc.add(column.toThrift());
+                }
+
                 // task sent to be
                 AgentBatchTask batchTask = new AgentBatchTask();
                 // count total replica num
@@ -256,7 +263,8 @@ public void process(DeleteStmt stmt) throws DdlException, QueryStateException {
                                         true, TPriority.NORMAL,
                                         TTaskType.REALTIME_PUSH,
                                         transactionId,
-                                        Catalog.getCurrentGlobalTransactionMgr().getTransactionIDGenerator().getNextTransactionId());
+                                        Catalog.getCurrentGlobalTransactionMgr().getTransactionIDGenerator().getNextTransactionId(),
+                                        columnsDesc);
                                 pushTask.setIsSchemaChanging(false);
                                 pushTask.setCountDownLatch(countDownLatch);
 

diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/LoadChecker.java b/fe/fe-core/src/main/java/org/apache/doris/load/LoadChecker.java
@@ -465,7 +465,7 @@ private Set<Long> submitPushTasks(LoadJob job, Database db) {
                                                                       needDecompress, job.getPriority(),
                                                                       TTaskType.REALTIME_PUSH,
                                                                       job.getTransactionId(),
-                                                                      Catalog.getCurrentGlobalTransactionMgr().getTransactionIDGenerator().getNextTransactionId());
+                                                                      Catalog.getCurrentGlobalTransactionMgr().getTransactionIDGenerator().getNextTransactionId(), null);
                                     pushTask.setIsSchemaChanging(autoLoadToTwoTablet);
                                     if (AgentTaskQueue.addTask(pushTask)) {
                                         batchTask.addTask(pushTask);

diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java
@@ -26,6 +26,7 @@
 import org.apache.doris.analysis.SlotRef;
 import org.apache.doris.common.MarkedCountDownLatch;
 import org.apache.doris.thrift.TBrokerScanRange;
+import org.apache.doris.thrift.TColumn;
 import org.apache.doris.thrift.TCondition;
 import org.apache.doris.thrift.TDescriptorTable;
 import org.apache.doris.thrift.TPriority;
@@ -69,11 +70,14 @@ public class PushTask extends AgentTask {
     private TBrokerScanRange tBrokerScanRange;
     private TDescriptorTable tDescriptorTable;
 
+    // for light schema change
+    private List<TColumn> columnsDesc = null;
+
     public PushTask(TResourceInfo resourceInfo, long backendId, long dbId, long tableId, long partitionId,
                     long indexId, long tabletId, long replicaId, int schemaHash, long version,
                     String filePath, long fileSize, int timeoutSecond, long loadJobId, TPushType pushType,
                     List<Predicate> conditions, boolean needDecompress, TPriority priority, TTaskType taskType,
-                    long transactionId, long signature) {
+                    long transactionId, long signature, List<TColumn> columnsDesc) {
         super(resourceInfo, backendId, taskType, dbId, tableId, partitionId, indexId, tabletId, signature);
         this.replicaId = replicaId;
         this.schemaHash = schemaHash;
@@ -92,16 +96,17 @@ public PushTask(TResourceInfo resourceInfo, long backendId, long dbId, long tabl
         this.transactionId = transactionId;
         this.tBrokerScanRange = null;
         this.tDescriptorTable = null;
+        this.columnsDesc = columnsDesc;
     }
 
     public PushTask(TResourceInfo resourceInfo, long backendId, long dbId, long tableId, long partitionId,
             long indexId, long tabletId, long replicaId, int schemaHash, long version,
             String filePath, long fileSize, int timeoutSecond, long loadJobId, TPushType pushType,
             List<Predicate> conditions, boolean needDecompress, TPriority priority) {
         this(resourceInfo, backendId, dbId, tableId, partitionId, indexId,
-                tabletId, replicaId, schemaHash, version, filePath,
-                fileSize, timeoutSecond, loadJobId, pushType, conditions, needDecompress,
-                priority, TTaskType.PUSH, -1, tableId);
+             tabletId, replicaId, schemaHash, version, filePath,
+             fileSize, timeoutSecond, loadJobId, pushType, conditions, needDecompress,
+             priority, TTaskType.PUSH, -1, tableId, null);
     }
 
     // for load v2 (SparkLoadJob)
@@ -110,9 +115,9 @@ public PushTask(long backendId, long dbId, long tableId, long partitionId, long
                     TPriority priority, long transactionId, long signature,
                     TBrokerScanRange tBrokerScanRange, TDescriptorTable tDescriptorTable) {
         this(null, backendId, dbId, tableId, partitionId, indexId,
-                tabletId, replicaId, schemaHash, -1, null,
-                0, timeoutSecond, loadJobId, pushType, null, false,
-                priority, TTaskType.REALTIME_PUSH, transactionId, signature);
+             tabletId, replicaId, schemaHash, -1, null,
+             0, timeoutSecond, loadJobId, pushType, null, false,
+             priority, TTaskType.REALTIME_PUSH, transactionId, signature, null);
         this.tBrokerScanRange = tBrokerScanRange;
         this.tDescriptorTable = tDescriptorTable;
     }
@@ -173,6 +178,7 @@ public TPushReq toThrift() {
                     tConditions.add(tCondition);
                 }
                 request.setDeleteConditions(tConditions);
+                request.setColumnsDesc(columnsDesc);
                 break;
             case LOAD_V2:
                 request.setBrokerScanRange(tBrokerScanRange);

diff --git a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
@@ -48,8 +48,6 @@
 
 import mockit.Expectations;
 import mockit.Injectable;
-import org.junit.Assert;
-import org.junit.Test;
 
 import java.util.List;
 import java.util.Map;

diff --git a/regression-test/suites/schema_change/test_delete_schema_change.sql b/regression-test/suites/schema_change/test_delete_schema_change.sql
@@ -27,11 +27,11 @@ INSERT INTO schema_change_delete_regression_test VALUES
 INSERT INTO schema_change_delete_regression_test VALUES
              (2, '2017-10-01', 'Beijing', 10, 1, '2020-01-03', '2020-01-03', '2020-01-03', 1, 32, 20);
 
-SELECT * FROM schema_change_delete_regression_test;
+SELECT * FROM schema_change_delete_regression_test order by user_id ASC, last_visit_date;
 
 ALTER table schema_change_delete_regression_test ADD COLUMN new_column INT default "1";
 
-SELECT * FROM schema_change_delete_regression_test;
+SELECT * FROM schema_change_delete_regression_test order by user_id DESC, last_visit_date;
 
 DELETE FROM schema_change_delete_regression_test where new_column = 1;