[feature-wip](array) remove array config and check array nested depth (#13428)

1. remove FE config `enable_array_type`
2. limit the nested depth of array in FE side.
3. Fix bug that when loading array from parquet, the decimal type is treated as bigint
4. Fix loading array from csv(vec-engine), handle null and "null"
5. Change the csv array loading behavior, if the array string format is invalid in csv, it will be converted to null. 
6. Remove `check_array_format()`, because it's logic is wrong and meaningless
7. Add stream load csv test cases and more parquet broker load tests

Author: morningman

be/src/runtime/decimalv2_value.cpp

@@ -354,7 +354,6 @@ int DecimalV2Value::parse_from_str(const char* decimal_str, int32_t length) {
 
     _value = StringParser::string_to_decimal<__int128>(decimal_str, length, PRECISION, SCALE,
                                                        &result);
-
     if (result == StringParser::PARSE_FAILURE) {
         error = E_DEC_BAD_NUM;
     }

be/src/vec/data_types/data_type_array.cpp

@@ -21,6 +21,7 @@
 #include "vec/data_types/data_type_array.h"
 
 #include "gen_cpp/data.pb.h"
+#include "util/stack_util.h"
 #include "vec/columns/column_array.h"
 #include "vec/columns/column_nullable.h"
 #include "vec/data_types/data_type_nullable.h"
@@ -175,15 +176,21 @@ std::string DataTypeArray::to_string(const IColumn& column, size_t row_num) cons
 }
 
 Status DataTypeArray::from_string(ReadBuffer& rb, IColumn* column) const {
+    DCHECK(!rb.eof());
     // only support one level now
     auto* array_column = assert_cast<ColumnArray*>(column);
     auto& offsets = array_column->get_offsets();
 
     IColumn& nested_column = array_column->get_data();
+    DCHECK(nested_column.is_nullable());
     if (*rb.position() != '[') {
         return Status::InvalidArgument("Array does not start with '[' character, found '{}'",
                                        *rb.position());
     }
+    if (*(rb.end() - 1) != ']') {
+        return Status::InvalidArgument("Array does not end with ']' character, found '{}'",
+                                       *(rb.end() - 1));
+    }
     ++rb.position();
     bool first = true;
     size_t size = 0;
@@ -210,13 +217,9 @@ Status DataTypeArray::from_string(ReadBuffer& rb, IColumn* column) const {
 
         // dispose the case of [123,,,]
         if (nested_str_len == 0) {
-            if (nested_column.is_nullable()) {
-                auto& nested_null_col = reinterpret_cast<ColumnNullable&>(nested_column);
-                nested_null_col.get_nested_column().insert_default();
-                nested_null_col.get_null_map_data().push_back(0);
-            } else {
-                nested_column.insert_default();
-            }
+            auto& nested_null_col = reinterpret_cast<ColumnNullable&>(nested_column);
+            nested_null_col.get_nested_column().insert_default();
+            nested_null_col.get_null_map_data().push_back(0);
             ++size;
             continue;
         }
@@ -236,19 +239,31 @@ Status DataTypeArray::from_string(ReadBuffer& rb, IColumn* column) const {
         }
 
         // dispose the case of ["123"] or ['123']
+        bool has_quota = false;
+        size_t tmp_len = nested_str_len;
         ReadBuffer read_buffer(rb.position(), nested_str_len);
         auto begin_char = *(rb.position() + begin_pos);
         auto end_char = *(rb.position() + end_pos);
         if (begin_char == end_char && (begin_char == '"' || begin_char == '\'')) {
             int64_t length = end_pos - begin_pos - 1;
             read_buffer = ReadBuffer(rb.position() + begin_pos + 1, (length > 0 ? length : 0));
+            tmp_len = (length > 0 ? length : 0);
+            has_quota = true;
         }
 
-        auto st = nested->from_string(read_buffer, &nested_column);
-        if (!st.ok()) {
-            // we should do revert if error
-            array_column->pop_back(size);
-            return st;
+        // handle null, need to distinguish null and "null"
+        if (!has_quota && tmp_len == 4 && strncmp(read_buffer.position(), "null", 4) == 0) {
+            // insert null
+            auto& nested_null_col = reinterpret_cast<ColumnNullable&>(nested_column);
+            nested_null_col.get_nested_column().insert_default();
+            nested_null_col.get_null_map_data().push_back(1);
+        } else {
+            auto st = nested->from_string(read_buffer, &nested_column);
+            if (!st.ok()) {
+                // we should do revert if error
+                array_column->pop_back(size);
+                return st;
+            }
         }
         rb.position() += nested_str_len;
         DCHECK_LE(rb.position(), rb.end());

be/src/vec/exec/format/csv/csv_reader.cpp

@@ -237,12 +237,6 @@ Status CsvReader::_fill_dest_columns(const Slice& line, std::vector<MutableColum
         return Status::OK();
     }
 
-    RETURN_IF_ERROR(_check_array_format(_split_values, &is_success));
-    if (UNLIKELY(!is_success)) {
-        // If not success, which means we met an invalid row, filter this row and return.
-        return Status::OK();
-    }
-
     // if _split_values.size > _file_slot_descs.size()
     // we only take the first few columns
     for (int i = 0; i < _file_slot_descs.size(); ++i) {

be/src/vec/exec/format/parquet/schema_desc.cpp

@@ -157,31 +157,35 @@ void FieldDescriptor::parse_physical_field(const tparquet::SchemaElement& physic
 
 TypeDescriptor FieldDescriptor::get_doris_type(const tparquet::SchemaElement& physical_schema) {
     TypeDescriptor type;
-    switch (physical_schema.type) {
-    case tparquet::Type::BOOLEAN:
-        type.type = TYPE_BOOLEAN;
-        return type;
-    case tparquet::Type::INT32:
-        type.type = TYPE_INT;
-        return type;
-    case tparquet::Type::INT64:
-    case tparquet::Type::INT96:
-        type.type = TYPE_BIGINT;
-        return type;
-    case tparquet::Type::FLOAT:
-        type.type = TYPE_FLOAT;
-        return type;
-    case tparquet::Type::DOUBLE:
-        type.type = TYPE_DOUBLE;
-        return type;
-    default:
-        break;
-    }
+    type.type = INVALID_TYPE;
     if (physical_schema.__isset.logicalType) {
         type = convert_to_doris_type(physical_schema.logicalType);
     } else if (physical_schema.__isset.converted_type) {
         type = convert_to_doris_type(physical_schema.converted_type);
     }
+    // use physical type instead
+    if (type.type == INVALID_TYPE) {
+        switch (physical_schema.type) {
+        case tparquet::Type::BOOLEAN:
+            type.type = TYPE_BOOLEAN;
+            return type;
+        case tparquet::Type::INT32:
+            type.type = TYPE_INT;
+            return type;
+        case tparquet::Type::INT64:
+        case tparquet::Type::INT96:
+            type.type = TYPE_BIGINT;
+            return type;
+        case tparquet::Type::FLOAT:
+            type.type = TYPE_FLOAT;
+            return type;
+        case tparquet::Type::DOUBLE:
+            type.type = TYPE_DOUBLE;
+            return type;
+        default:
+            break;
+        }
+    }
     return type;
 }
 
@@ -214,7 +218,7 @@ TypeDescriptor FieldDescriptor::convert_to_doris_type(tparquet::LogicalType logi
     } else if (logicalType.__isset.TIMESTAMP) {
         type.type = TYPE_DATETIMEV2;
     } else {
-        LOG(WARNING) << "Not supported parquet LogicalType";
+        type.type = INVALID_TYPE;
     }
     return type;
 }
@@ -253,6 +257,7 @@ TypeDescriptor FieldDescriptor::convert_to_doris_type(tparquet::ConvertedType::t
         break;
     default:
         LOG(WARNING) << "Not supported parquet ConvertedType: " << convertedType;
+        type = INVALID_TYPE;
         break;
     }
     return type;

be/src/vec/exec/scan/vfile_scanner.cpp

@@ -509,7 +509,7 @@ Status VFileScanner::_get_next_reader() {
         _name_to_col_type.clear();
         _missing_cols.clear();
         _cur_reader->get_columns(&_name_to_col_type, &_missing_cols);
-        if (!_missing_cols.empty() && _is_load && VLOG_NOTICE_IS_ON) {
+        if (VLOG_NOTICE_IS_ON && !_missing_cols.empty() && _is_load) {
             fmt::memory_buffer col_buf;
             for (auto& col : _missing_cols) {
                 fmt::format_to(col_buf, " {}", col);

be/src/vec/exec/scan/vscanner.cpp

@@ -120,9 +120,6 @@ Status VScanner::close(RuntimeState* state) {
 }
 
 void VScanner::_update_counters_before_close() {
-    LOG(INFO) << "cmy _update_counters_before_close: _counter.num_rows_filtered: "
-              << _counter.num_rows_filtered
-              << ", _counter.num_rows_unselected: " << _counter.num_rows_unselected;
     if (!_state->enable_profile() && !_is_load) return;
     COUNTER_UPDATE(_parent->_rows_read_counter, _num_rows_read);
     // Update stats for load

be/src/vec/exec/vbroker_scanner.cpp

@@ -91,9 +91,10 @@ Status VBrokerScanner::_fill_dest_columns(const Slice& line,
         return Status::OK();
     }
 
-    if (!check_array_format(_split_values)) {
-        return Status::OK();
-    }
+    // This check is meaningless, should be removed
+    // if (!check_array_format(_split_values)) {
+    //     return Status::OK();
+    // }
 
     int idx = 0;
     for (int i = 0; i < _split_values.size(); ++i) {

be/src/vec/functions/function_cast.h

@@ -320,10 +320,12 @@ struct ConvertImplGenericFromString {
                     check_and_get_column<StringColumnType>(&col_from)) {
             auto col_to = data_type_to->create_column();
 
-            //IColumn & col_to = *res;
             size_t size = col_from.size();
             col_to->reserve(size);
 
+            ColumnUInt8::MutablePtr col_null_map_to = ColumnUInt8::create(size);
+            ColumnUInt8::Container* vec_null_map_to = &col_null_map_to->get_data();
+
             for (size_t i = 0; i < size; ++i) {
                 const auto& val = col_from_string->get_data_at(i);
                 // Note: here we should handle the null element
@@ -332,9 +334,15 @@ struct ConvertImplGenericFromString {
                     continue;
                 }
                 ReadBuffer read_buffer((char*)(val.data), val.size);
-                RETURN_IF_ERROR(data_type_to->from_string(read_buffer, col_to));
+                Status st = data_type_to->from_string(read_buffer, col_to);
+                // if parsing failed, will return null
+                (*vec_null_map_to)[i] = !st.ok();
+                if (!st.ok()) {
+                    col_to->insert_default();
+                }
             }
-            block.replace_by_position(result, std::move(col_to));
+            block.get_by_position(result).column =
+                    ColumnNullable::create(std::move(col_to), std::move(col_null_map_to));
         } else {
             return Status::RuntimeError(
                     "Illegal column {} of first argument of conversion function from string",
@@ -870,8 +878,9 @@ struct ConvertThroughParsing {
         if constexpr (IsDataTypeDecimal<ToDataType>) {
             UInt32 scale = additions;
             col_to = ColVecTo::create(size, scale);
-        } else
+        } else {
             col_to = ColVecTo::create(size);
+        }
 
         typename ColVecTo::Container& vec_to = col_to->get_data();
 

be/src/vec/io/reader_buffer.h

@@ -35,6 +35,8 @@ class ReadBuffer {
 
     size_t count() { return _end - _start; }
 
+    std::string to_string() { return std::string(_start, (_end - _start)); }
+
 private:
     char* _start;
     char* _end;

docs/zh-CN/docs/sql-manual/sql-reference/Data-Types/ARRAY.md

@@ -28,7 +28,7 @@ under the License.
 
 ### description
 
-ARRAY\<T\>
+`ARRAY<T>`
 
 由T类型元素组成的数组，不能作为key列使用。目前支持在Duplicate模型的表中使用。
 
@@ -74,7 +74,7 @@ PROPERTIES (
 mysql> INSERT INTO `array_test` VALUES (1, [1,2,3,4,5]);
 mysql> INSERT INTO `array_test` VALUES (2, array(6,7,8)), (3, array()), (4, null);
 ```
-注意：以上sql仅在非向量化场景下，支持array(）函数，向量化场景不支持。
+注意：以上sql仅在非向量化场景下，支持 array() 函数，向量化场景不支持。
 
 查询数据示例：
 

fe/check/checkstyle/checkstyle.xml

@@ -57,10 +57,6 @@ under the License.
         <property name="lineSeparator" value="lf"/>
     </module>
 
-    <module name="RegexpSingleline">
-        <property name="format" value="&gt;&gt;&gt;&gt;&gt;&gt;&gt;"/>
-        <property name="message" value="Merge conflicts unresolved."/>
-    </module>
     <module name="RegexpSingleline">
         <property name="format" value="&lt;&lt;&lt;&lt;&lt;&lt;&lt;"/>
         <property name="message" value="Merge conflicts unresolved."/>

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

@@ -267,7 +267,7 @@ public void analyze(boolean isOlap) throws AnalysisException {
         if (type.getPrimitiveType() == PrimitiveType.ARRAY) {
             if (isKey()) {
                 throw new AnalysisException("Array can only be used in the non-key column of"
-                    + " the duplicate table at present.");
+                        + " the duplicate table at present.");
             }
             if (defaultValue.isSet && defaultValue != DefaultValue.NULL_DEFAULT_VALUE) {
                 throw new AnalysisException("Array type column default value only support null");

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

@@ -328,6 +328,9 @@ public void analyze(Analyzer analyzer) throws UserException {
                         if (columnDef.getType().getPrimitiveType() == PrimitiveType.JSONB) {
                             break;
                         }
+                        if (columnDef.getType().isCollectionType()) {
+                            break;
+                        }
                         if (columnDef.getType().getPrimitiveType() == PrimitiveType.VARCHAR) {
                             keysColumnNames.add(columnDef.getName());
                             break;
@@ -393,9 +396,6 @@ public void analyze(Analyzer analyzer) throws UserException {
             columnDef.analyze(engineName.equals("olap"));
 
             if (columnDef.getType().isArrayType()) {
-                if (!Config.enable_array_type) {
-                    throw new AnalysisException("Please open enable_array_type config before use Array.");
-                }
                 if (columnDef.getAggregateType() != null && columnDef.getAggregateType() != AggregateType.NONE) {
                     throw new AnalysisException("Array column can't support aggregation "
                             + columnDef.getAggregateType());

fe/fe-core/src/main/java/org/apache/doris/catalog/ArrayType.java

@@ -33,6 +33,8 @@
  */
 public class ArrayType extends Type {
 
+    public static final int MAX_NESTED_DEPTH = 9;
+
     @SerializedName(value = "itemType")
     private Type itemType;
 

fe/fe-core/src/main/java/org/apache/doris/catalog/Type.java

@@ -46,12 +46,8 @@
  * as abstract methods that subclasses must implement.
  */
 public abstract class Type {
-    // Maximum nesting depth of a type. This limit was determined experimentally by
-    // org.apache.doris.rewrite.FoldConstantsRule.apply generating and scanning
-    // deeply nested Parquet and Avro files. In those experiments, we exceeded
-    // the stack space in the scanner (which uses recursion for dealing with
-    // nested types) at a nesting depth between 200 and 300 (200 worked, 300 crashed).
-    public static int MAX_NESTING_DEPTH = 2;
+    // Currently only support Array type with max 9 depths.
+    public static int MAX_NESTING_DEPTH = 9;
 
     // Static constant types for scalar types that don't require additional information.
     public static final ScalarType INVALID = new ScalarType(PrimitiveType.INVALID_TYPE);
@@ -488,7 +484,7 @@ public static boolean canCastTo(Type t1, Type t2) {
         } else if (t1.isArrayType() && t2.isArrayType()) {
             return ArrayType.canCastTo((ArrayType) t1, (ArrayType) t2);
         }
-        return t1.isNull() || t1.getPrimitiveType() == PrimitiveType.VARCHAR;
+        return t1.isNull() || t1.getPrimitiveType().isCharFamily();
     }
 
     /**
@@ -612,7 +608,7 @@ public boolean exceedsMaxNestingDepth() {
      * MAP<STRING,STRUCT<f1:INT>> --> 3
      */
     private boolean exceedsMaxNestingDepth(int d) {
-        if (d >= MAX_NESTING_DEPTH) {
+        if (d > MAX_NESTING_DEPTH) {
             return true;
         }
         if (isStructType()) {
@@ -623,7 +619,9 @@ private boolean exceedsMaxNestingDepth(int d) {
                 }
             }
         } else if (isArrayType()) {
-            return false;
+            ArrayType arrayType = (ArrayType) this;
+            Type itemType = arrayType.getItemType();
+            return itemType.exceedsMaxNestingDepth(d + 1);
         } else if (isMultiRowType()) {
             MultiRowType multiRowType = (MultiRowType) this;
             return multiRowType.getItemType().exceedsMaxNestingDepth(d + 1);

fe/fe-core/src/test/java/org/apache/doris/analysis/ColumnDefTest.java

@@ -25,7 +25,6 @@
 import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.qe.ConnectContext;
 
 import mockit.Mock;
@@ -47,7 +46,6 @@ public void setUp() {
         stringCol = new TypeDef(ScalarType.createChar(10));
         floatCol = new TypeDef(ScalarType.createType(PrimitiveType.FLOAT));
         booleanCol = new TypeDef(ScalarType.createType(PrimitiveType.BOOLEAN));
-        Config.enable_array_type = true;
 
         ctx = new ConnectContext(null);
         new MockUp<ConnectContext>() {