Merge remote-tracking branch 'origin/perf-base-image' into perf-base-image

catpineapple · catpineapple · commit 587f82b2ee60 · 2025-04-29T15:30:25.000+08:00
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
diff --git a/fe/fe-core/src/main/java/org/apache/doris/binlog/AddPartitionRecord.java b/fe/fe-core/src/main/java/org/apache/doris/binlog/AddPartitionRecord.java
@@ -90,8 +90,12 @@ public AddPartitionRecord(long commitSeq, PartitionPersistInfo partitionPersistI
             sb.append("\");");
         } else if (!this.listPartitionItem.equals(ListPartitionItem.DUMMY_ITEM)) {
             // list
-            sb.append("VALUES IN ");
-            sb.append(((ListPartitionItem) listPartitionItem).toSql());
+            String partitionSql = ((ListPartitionItem) listPartitionItem).toSql();
+            // the partition may default partition
+            if (!partitionSql.isEmpty()) {
+                sb.append("VALUES IN ");
+                sb.append(partitionSql);
+            }
             sb.append(" (\"version_info\" = \"");
             sb.append(partition.getVisibleVersion());
             sb.append("\");");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionInfo.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionInfo.java
@@ -207,21 +207,27 @@ public String toSql(OlapTable table, List<Long> partitionId) {
             String partitionName = partition.getName();
             List<PartitionKey> partitionKeys = entry.getValue().getItems();
 
-            sb.append("PARTITION ").append(partitionName).append(" VALUES IN ");
-            sb.append("(");
+            sb.append("PARTITION ").append(partitionName);
+            StringBuilder partitionKeysBuilder = new StringBuilder();
             int idxInternal = 0;
             for (PartitionKey partitionKey : partitionKeys) {
                 String partitionKeyStr = partitionKey.toSql();
-                if (!isMultiColumnPartition) {
+                if (!isMultiColumnPartition && !partitionKeyStr.isEmpty()) {
                     partitionKeyStr = partitionKeyStr.substring(1, partitionKeyStr.length() - 1);
                 }
-                sb.append(partitionKeyStr);
+                partitionKeysBuilder.append(partitionKeyStr);
                 if (partitionKeys.size() > 1 && idxInternal != partitionKeys.size() - 1) {
-                    sb.append(",");
+                    partitionKeysBuilder.append(",");
                 }
                 idxInternal++;
             }
-            sb.append(")");
+
+            // length == 0 means it is a default partition
+            if (partitionKeysBuilder.length() > 0) {
+                sb.append(" VALUES IN ").append("(");
+                sb.append(partitionKeysBuilder.toString());
+                sb.append(")");
+            }
 
             if (!"".equals(getStoragePolicy(entry.getKey()))) {
                 sb.append("(\"storage_policy\" = \"").append(getStoragePolicy(entry.getKey())).append("\")");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionItem.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ListPartitionItem.java
@@ -188,6 +188,9 @@ public String toString() {
     }
 
     public String toSql() {
+        if (isDefaultPartition) {
+            return "";
+        }
         StringBuilder sb = new StringBuilder();
         sb.append("(");
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionKey.java
@@ -40,6 +40,7 @@
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
+import com.google.gson.Gson;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonDeserializationContext;
 import com.google.gson.JsonDeserializer;
@@ -69,6 +70,7 @@ public class PartitionKey implements Comparable<PartitionKey>, Writable {
     private List<String> originHiveKeys;
     @SerializedName("ts")
     private List<PrimitiveType> types;
+    // the isD not serialize before because of Gson using PartitionKeySerializer
     @SerializedName("isD")
     private boolean isDefaultListPartitionKey = false;
 
@@ -366,6 +368,9 @@ public PartitionKey successor() throws AnalysisException {
 
     // return: ("100", "200", "300")
     public String toSql() {
+        if (isDefaultListPartitionKey) {
+            return "";
+        }
         StringBuilder sb = new StringBuilder("(");
         int i = 0;
         for (LiteralExpr expr : keys) {
@@ -594,8 +599,13 @@ public JsonElement serialize(PartitionKey partitionKey, java.lang.reflect.Type r
                 jsonArray.add(typeAndKey);
             }
 
-            // for compatibility in the future
-            jsonArray.add(new JsonPrimitive("unused"));
+            // use extend as a json to record some information
+            JsonArray extend = new JsonArray();
+            extend.add(context.serialize(partitionKey.isDefaultListPartitionKey()));
+
+            // for compatibility in the future add item before unused
+            extend.add("unused");
+            jsonArray.add(new JsonPrimitive(GsonUtils.GSON.toJson(extend)));
 
             return jsonArray;
         }
@@ -640,8 +650,18 @@ public PartitionKey deserialize(JsonElement json, java.lang.reflect.Type typeOfT
                     partitionKey.types.add(type);
                     partitionKey.keys.add(key);
                 }
+                JsonPrimitive extend = jsonArray.get(jsonArray.size() - 1).getAsJsonPrimitive();
+                String extendStr = extend.getAsString();
+                // for compatibility, extend takes up the previous "unused" position
+                // so the last element needs to be checked here, if it is unused ignore it
+                if (!extendStr.equals("unused")) {
+                    // parse extend record
+                    Gson gson = new Gson();
+                    JsonArray pExtend = gson.fromJson(extendStr, JsonArray.class);
+                    partitionKey.setDefaultListPartition(pExtend.get(0).getAsBoolean());
+                    // ignore the last element
+                }
 
-                // ignore the last element
                 return partitionKey;
             }
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/AvroFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/AvroFileFormatProperties.java
@@ -36,8 +36,7 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo
     }
 
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java
@@ -134,8 +134,9 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo
     }
 
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
+        sinkOptions.setColumnSeparator(columnSeparator);
+        sinkOptions.setLineDelimiter(lineDelimiter);
     }
 
     // The method `analyzeFileFormatProperties` must have been called once before this method
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/FileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/FileFormatProperties.java
@@ -62,7 +62,7 @@ public abstract void analyzeFileFormatProperties(
      * generate TResultFileSinkOptions according to the properties of specified file format
      * You must call method `analyzeFileFormatProperties` once before calling method `toTResultFileSinkOptions`
      */
-    public abstract TResultFileSinkOptions toTResultFileSinkOptions();
+    public abstract void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions);
 
     /**
      * generate TFileAttributes according to the properties of specified file format
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/JsonFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/JsonFileFormatProperties.java
@@ -73,8 +73,7 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo
     }
 
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/OrcFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/OrcFileFormatProperties.java
@@ -24,9 +24,20 @@
 import org.apache.doris.thrift.TFileTextScanRangeParams;
 import org.apache.doris.thrift.TResultFileSinkOptions;
 
+import com.google.common.collect.Maps;
+
 import java.util.Map;
 
 public class OrcFileFormatProperties extends FileFormatProperties {
+    public static final Map<String, TFileCompressType> ORC_COMPRESSION_TYPE_MAP = Maps.newHashMap();
+
+    static {
+        ORC_COMPRESSION_TYPE_MAP.put("plain", TFileCompressType.PLAIN);
+        ORC_COMPRESSION_TYPE_MAP.put("snappy", TFileCompressType.SNAPPYBLOCK);
+        ORC_COMPRESSION_TYPE_MAP.put("zlib", TFileCompressType.ZLIB);
+        ORC_COMPRESSION_TYPE_MAP.put("zstd", TFileCompressType.ZSTD);
+    }
+
     private TFileCompressType orcCompressionType = TFileCompressType.ZLIB;
 
     public OrcFileFormatProperties() {
@@ -36,11 +47,26 @@ public OrcFileFormatProperties() {
     @Override
     public void analyzeFileFormatProperties(Map<String, String> formatProperties, boolean isRemoveOriginProperty)
             throws AnalysisException {
+        // get compression type
+        // save compress type
+        if (formatProperties.containsKey(PROP_COMPRESS_TYPE)) {
+            if (ORC_COMPRESSION_TYPE_MAP.containsKey(
+                    formatProperties.get(PROP_COMPRESS_TYPE).toLowerCase())) {
+                this.orcCompressionType = ORC_COMPRESSION_TYPE_MAP.get(
+                        formatProperties.get(PROP_COMPRESS_TYPE).toLowerCase());
+                formatProperties.remove(PROP_COMPRESS_TYPE);
+            } else {
+                throw new AnalysisException("orc compression type ["
+                        + formatProperties.get(PROP_COMPRESS_TYPE) + "] is invalid,"
+                        + " please choose one among ZLIB, SNAPPY, ZSTD or PLAIN");
+            }
+        }
     }
 
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
+        sinkOptions.setOrcCompressionType(orcCompressionType);
+        sinkOptions.setOrcWriterVersion(1);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/ParquetFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/ParquetFileFormatProperties.java
@@ -25,15 +25,41 @@
 import org.apache.doris.thrift.TParquetVersion;
 import org.apache.doris.thrift.TResultFileSinkOptions;
 
+import com.google.common.collect.Maps;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Map.Entry;
 
 public class ParquetFileFormatProperties extends FileFormatProperties {
     public static final String PARQUET_DISABLE_DICTIONARY = "disable_dictionary";
-    public static final TParquetVersion parquetVersion = TParquetVersion.PARQUET_1_0;
     public static final String PARQUET_VERSION = "version";
+    public static final String PARQUET_PROP_PREFIX = "parquet.";
+
+    public static final Logger LOG = LogManager.getLogger(ParquetFileFormatProperties.class);
+    public static final Map<String, TParquetCompressionType> PARQUET_COMPRESSION_TYPE_MAP = Maps.newHashMap();
+    public static final Map<String, TParquetVersion> PARQUET_VERSION_MAP = Maps.newHashMap();
+
+    static {
+        PARQUET_COMPRESSION_TYPE_MAP.put("snappy", TParquetCompressionType.SNAPPY);
+        PARQUET_COMPRESSION_TYPE_MAP.put("gzip", TParquetCompressionType.GZIP);
+        PARQUET_COMPRESSION_TYPE_MAP.put("brotli", TParquetCompressionType.BROTLI);
+        PARQUET_COMPRESSION_TYPE_MAP.put("zstd", TParquetCompressionType.ZSTD);
+        PARQUET_COMPRESSION_TYPE_MAP.put("lz4", TParquetCompressionType.LZ4);
+        // arrow do not support lzo and bz2 compression type.
+        // PARQUET_COMPRESSION_TYPE_MAP.put("lzo", TParquetCompressionType.LZO);
+        // PARQUET_COMPRESSION_TYPE_MAP.put("bz2", TParquetCompressionType.BZ2);
+        PARQUET_COMPRESSION_TYPE_MAP.put("plain", TParquetCompressionType.UNCOMPRESSED);
+
+        PARQUET_VERSION_MAP.put("v1", TParquetVersion.PARQUET_1_0);
+        PARQUET_VERSION_MAP.put("latest", TParquetVersion.PARQUET_2_LATEST);
+    }
 
     private TParquetCompressionType parquetCompressionType = TParquetCompressionType.SNAPPY;
     private boolean parquetDisableDictionary = false;
+    private TParquetVersion parquetVersion = TParquetVersion.PARQUET_1_0;
 
     public ParquetFileFormatProperties() {
         super(TFileFormatType.FORMAT_PARQUET);
@@ -42,11 +68,47 @@ public ParquetFileFormatProperties() {
     @Override
     public void analyzeFileFormatProperties(Map<String, String> formatProperties, boolean isRemoveOriginProperty)
             throws AnalysisException {
+        // save compress type
+        if (formatProperties.containsKey(PROP_COMPRESS_TYPE)) {
+            if (PARQUET_COMPRESSION_TYPE_MAP.containsKey(formatProperties.get(PROP_COMPRESS_TYPE)
+                    .toLowerCase())) {
+                this.parquetCompressionType = PARQUET_COMPRESSION_TYPE_MAP.get(
+                        formatProperties.get(PROP_COMPRESS_TYPE).toLowerCase());
+                formatProperties.remove(PROP_COMPRESS_TYPE);
+            } else {
+                throw new AnalysisException("parquet compression type ["
+                        + formatProperties.get(PROP_COMPRESS_TYPE)
+                        + "] is invalid, please choose one among SNAPPY, GZIP, BROTLI, ZSTD, LZ4, LZO, BZ2 or PLAIN");
+            }
+        }
+
+        // save all parquet prefix property
+        Iterator<Entry<String, String>> iter = formatProperties.entrySet().iterator();
+        while (iter.hasNext()) {
+            Map.Entry<String, String> entry = iter.next();
+            if (entry.getKey().startsWith(PARQUET_PROP_PREFIX)) {
+                iter.remove();
+                if (entry.getKey().substring(PARQUET_PROP_PREFIX.length())
+                        .equals(PARQUET_DISABLE_DICTIONARY)) {
+                    this.parquetDisableDictionary = Boolean.valueOf(entry.getValue());
+                } else if (entry.getKey().substring(PARQUET_PROP_PREFIX.length())
+                        .equals(PARQUET_VERSION)) {
+                    if (PARQUET_VERSION_MAP.containsKey(entry.getValue())) {
+                        this.parquetVersion = PARQUET_VERSION_MAP.get(entry.getValue());
+                    } else {
+                        LOG.debug("not set parquet version type or is invalid, set default to PARQUET_1.0 version.");
+                    }
+                }
+            }
+        }
     }
 
+
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
+        sinkOptions.setParquetCompressionType(parquetCompressionType);
+        sinkOptions.setParquetDisableDictionary(parquetDisableDictionary);
+        sinkOptions.setParquetVersion(parquetVersion);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/WalFileFormatProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/WalFileFormatProperties.java
@@ -31,8 +31,7 @@ public WalFileFormatProperties() {
     }
 
     @Override
-    public TResultFileSinkOptions toTResultFileSinkOptions() {
-        return null;
+    public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ResultFileSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ResultFileSink.java
@@ -25,6 +25,7 @@
 import org.apache.doris.analysis.TupleId;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.util.FileFormatConstants;
+import org.apache.doris.common.util.Util;
 import org.apache.doris.thrift.TDataSink;
 import org.apache.doris.thrift.TDataSinkType;
 import org.apache.doris.thrift.TExplainLevel;
@@ -69,11 +70,13 @@ private String genNames(ArrayList<String> headerNames, String columnSeparator, S
 
     public ResultFileSink(PlanNodeId exchNodeId, OutFileClause outFileClause, ArrayList<String> labels) {
         this(exchNodeId, outFileClause);
-        if (outFileClause.getHeaderType().equals(FileFormatConstants.FORMAT_CSV_WITH_NAMES)
-                || outFileClause.getHeaderType().equals(FileFormatConstants.FORMAT_CSV_WITH_NAMES_AND_TYPES)) {
-            header = genNames(labels, outFileClause.getColumnSeparator(), outFileClause.getLineDelimiter());
+        if (Util.isCsvFormat(outFileClause.getFileFormatType())) {
+            if (outFileClause.getHeaderType().equals(FileFormatConstants.FORMAT_CSV_WITH_NAMES)
+                    || outFileClause.getHeaderType().equals(FileFormatConstants.FORMAT_CSV_WITH_NAMES_AND_TYPES)) {
+                header = genNames(labels, outFileClause.getColumnSeparator(), outFileClause.getLineDelimiter());
+            }
+            headerType = outFileClause.getHeaderType();
         }
-        headerType = outFileClause.getHeaderType();
     }
 
     public String getBrokerName() {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/PartitionKeyTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/PartitionKeyTest.java
@@ -219,7 +219,9 @@ public void testSerialization() throws Exception {
         FakeEnv.setMetaVersion(FeConstants.meta_version);
 
         // 1. Write objects to file
-        Path path = Files.createFile(Paths.get("./keyRangePartition"));
+        Path dir = Paths.get("./keyRangePartition");
+        Files.deleteIfExists(dir);
+        Path path = Files.createFile(dir);
         DataOutputStream dos = new DataOutputStream(Files.newOutputStream(path));
 
         PartitionKey keyEmpty = new PartitionKey();
@@ -251,6 +253,13 @@ public void testSerialization() throws Exception {
         PartitionKey key = PartitionKey.createPartitionKey(keys, columns);
         key.write(dos);
 
+        keys.clear();
+        List<Type> types = new ArrayList<Type>();
+        types.add(ScalarType.createType(PrimitiveType.INT));
+        PartitionKey defaultKey = PartitionKey.createListPartitionKeyWithTypes(keys, types, false);
+        Assert.assertTrue(defaultKey.isDefaultListPartitionKey());
+        defaultKey.write(dos);
+
         dos.flush();
         dos.close();
 
@@ -264,6 +273,10 @@ public void testSerialization() throws Exception {
         Assert.assertEquals(key, key);
         Assert.assertNotEquals(key, this);
 
+        PartitionKey rDefaultKey = PartitionKey.read(dis);
+        Assert.assertEquals(defaultKey, rDefaultKey);
+        Assert.assertTrue(rDefaultKey.isDefaultListPartitionKey());
+
         // 3. delete files
         dis.close();
         Files.deleteIfExists(path);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/fileformat/OrcFileFormatPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/fileformat/OrcFileFormatPropertiesTest.java
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/fileformat/ParquetFileFormatPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/fileformat/ParquetFileFormatPropertiesTest.java
diff --git a/regression-test/suites/partition_p0/list_partition/test_list_default_partition_show_create.groovy b/regression-test/suites/partition_p0/list_partition/test_list_default_partition_show_create.groovy

Original file line number	Diff line number	Diff line change
`@@ -188,6 +188,9 @@ public String toString() {`
`188`	`188`	`}`
`189`	`189`
`190`	`190`	`public String toSql() {`
	`191`	`+ if (isDefaultPartition) {`
	`192`	`+ return "";`
	`193`	`+ }`
`191`	`194`	`StringBuilder sb = new StringBuilder();`
`192`	`195`	`sb.append("(");`
`193`	`196`
Original file line number	Diff line number	Diff line change
`@@ -36,8 +36,7 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`@Override`
`39`		`- public TResultFileSinkOptions toTResultFileSinkOptions() {`
`40`		`- return null;`
	`39`	`+ public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {`
`41`	`40`	`}`
`42`	`41`
`43`	`42`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -134,8 +134,9 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo`
`134`	`134`	`}`
`135`	`135`
`136`	`136`	`@Override`
`137`		`- public TResultFileSinkOptions toTResultFileSinkOptions() {`
`138`		`- return null;`
	`137`	`+ public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {`
	`138`	`+ sinkOptions.setColumnSeparator(columnSeparator);`
	`139`	`+ sinkOptions.setLineDelimiter(lineDelimiter);`
`139`	`140`	`}`
`140`	`141`
`141`	`142`	// The method `analyzeFileFormatProperties` must have been called once before this method
Original file line number	Diff line number	Diff line change
`@@ -73,8 +73,7 @@ public void analyzeFileFormatProperties(Map<String, String> formatProperties, bo`
`73`	`73`	`}`
`74`	`74`
`75`	`75`	`@Override`
`76`		`- public TResultFileSinkOptions toTResultFileSinkOptions() {`
`77`		`- return null;`
	`76`	`+ public void fullTResultFileSinkOptions(TResultFileSinkOptions sinkOptions) {`
`78`	`77`	`}`
`79`	`78`
`80`	`79`	`@Override`