Data export function, add export to specify certain columns

docs/zh-CN/administrator-guide/export-manual.md

@@ -111,6 +111,7 @@ TO "hdfs://host/path/to/export/"
 PROPERTIES
 (
     "column_separator"=",",
+    "columns":"col1,col2"
     "exec_mem_limit"="2147483648",
     "timeout" = "3600"
 )
@@ -122,6 +123,7 @@ WITH BROKER "hdfs"
 ```
 
 * `column_separator`：列分隔符。默认为 `\t`。支持不可见字符，比如 '\x07'。
+* columns：要导出的列，使用英文状态逗号隔开，如果不填这个参数默认是导出表的所有列
 * `line_delimiter`：行分隔符。默认为 `\n`。支持不可见字符，比如 '\x07'。
 * `exec_mem_limit`： 表示 Export 作业中，一个查询计划在单个 BE 上的内存使用限制。默认 2GB。单位字节。
 * `timeout`：作业超时时间。默认 2小时。单位秒。

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

@@ -60,6 +60,8 @@ public class ExportStmt extends StatementBase {
 
     private static final String DEFAULT_COLUMN_SEPARATOR = "\t";
     private static final String DEFAULT_LINE_DELIMITER = "\n";
+    private static final String DEFAULT_COLUMNS = "";
+
 
     private TableName tblName;
     private List<String> partitions;
@@ -69,6 +71,7 @@ public class ExportStmt extends StatementBase {
     private Map<String, String> properties = Maps.newHashMap();
     private String columnSeparator;
     private String lineDelimiter;
+    private String columns ;
 
     private TableRef tableRef;
 
@@ -83,6 +86,11 @@ public ExportStmt(TableRef tableRef, Expr whereExpr, String path,
         this.brokerDesc = brokerDesc;
         this.columnSeparator = DEFAULT_COLUMN_SEPARATOR;
         this.lineDelimiter = DEFAULT_LINE_DELIMITER;
+        this.columns = DEFAULT_COLUMNS;
+    }
+
+    public String getColumns() {
+        return columns;
     }
 
     public TableName getTblName() {
@@ -264,6 +272,7 @@ private void checkProperties(Map<String, String> properties) throws UserExceptio
                 properties, ExportStmt.DEFAULT_COLUMN_SEPARATOR));
         this.lineDelimiter = Separator.convertSeparator(PropertyAnalyzer.analyzeLineDelimiter(
                 properties, ExportStmt.DEFAULT_LINE_DELIMITER));
+       this.columns = properties.get(LoadStmt.KEY_IN_PARAM_COLUMNS);
         // exec_mem_limit
         if (properties.containsKey(LoadStmt.EXEC_MEM_LIMIT)) {
             try {

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

@@ -82,6 +82,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import com.google.common.base.Splitter;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.logging.log4j.LogManager;
@@ -167,6 +168,10 @@ public enum JobState {
     private OriginStatement origStmt;
     protected Map<String, String> sessionVariables = Maps.newHashMap();
 
+    private List<String> exportColumns = Lists.newArrayList();
+    private String columns ;
+
+
     public ExportJob() {
         this.id = -1;
         this.dbId = -1;
@@ -182,6 +187,7 @@ public ExportJob() {
         this.exportPath = "";
         this.columnSeparator = "\t";
         this.lineDelimiter = "\n";
+        this.columns = "";
     }
 
     public ExportJob(long jobId) {
@@ -211,7 +217,11 @@ public void setJob(ExportStmt stmt) throws UserException {
         this.partitions = stmt.getPartitions();
 
         this.exportTable = db.getTable(stmt.getTblName().getTbl());
-
+        this.columns = stmt.getColumns();
+        if (!Strings.isNullOrEmpty(this.columns)) {
+            Splitter split = Splitter.on(',').trimResults().omitEmptyStrings();
+            this.exportColumns = split.splitToList(stmt.getColumns().toLowerCase());
+        }
         exportTable.readLock();
         try {
             this.dbId = db.getId();
@@ -259,10 +269,18 @@ private void registerToDesc() {
         exportTupleDesc.setTable(exportTable);
         exportTupleDesc.setRef(tableRef);
         for (Column col : exportTable.getBaseSchema()) {
-            SlotDescriptor slot = desc.addSlotDescriptor(exportTupleDesc);
-            slot.setIsMaterialized(true);
-            slot.setColumn(col);
-            slot.setIsNullable(col.isAllowNull());
+            String colName = col.getName().toLowerCase();
+            if (!this.exportColumns.isEmpty() && this.exportColumns.contains(colName)) {
+                SlotDescriptor slot = desc.addSlotDescriptor(exportTupleDesc);
+                slot.setIsMaterialized(true);
+                slot.setColumn(col);
+                slot.setIsNullable(col.isAllowNull());
+            } else {
+                SlotDescriptor slot = desc.addSlotDescriptor(exportTupleDesc);
+                slot.setIsMaterialized(true);
+                slot.setColumn(col);
+                slot.setIsNullable(col.isAllowNull());
+            }
         }
         desc.computeMemLayout();
     }
@@ -447,6 +465,10 @@ private void genCoordinators(List<PlanFragment> fragments, List<ScanNode> nodes)
         LOG.info("create {} coordinators for export job: {}", coordList.size(), id);
     }
 
+    public String getColumns() {
+        return columns;
+    }
+
     public long getId() {
         return id;
     }
@@ -740,7 +762,11 @@ public void readFields(DataInput in) throws IOException {
                 this.properties.put(propertyKey, propertyValue);
             }
         }
-
+        this.columns = this.properties.get(LoadStmt.KEY_IN_PARAM_COLUMNS);
+        if (!Strings.isNullOrEmpty(this.columns)) {
+            Splitter split = Splitter.on(',').trimResults().omitEmptyStrings();
+            this.exportColumns = split.splitToList(this.columns);
+        }
         boolean hasPartition = in.readBoolean();
         if (hasPartition) {
             partitions = Lists.newArrayList();
@@ -782,7 +808,7 @@ public void readFields(DataInput in) throws IOException {
             String value = Text.readString(in);
             sessionVariables.put(key, value);
         }
-        
+
         if (origStmt.originStmt.isEmpty()) {
             return;
         }

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

@@ -193,6 +193,7 @@ public List<List<String>> getExportJobInfosByIdOrState(
                 infoMap.put("column separator", job.getColumnSeparator());
                 infoMap.put("line delimiter", job.getLineDelimiter());
                 infoMap.put("exec mem limit", job.getExecMemLimit());
+                infoMap.put("columns", job.getColumns());
                 infoMap.put("coord num", job.getCoordList().size());
                 infoMap.put("tablet num", job.getTabletLocations() == null ? -1 : job.getTabletLocations().size());
                 jobInfo.add(new Gson().toJson(infoMap));