Merge pull request ClickHouse#62210 from bigo-sg/json_format_early_skip

Improve `JSONEachRowRowInputFormat` by skipping all remaining fields when all required fields are read

Author: antaljanosbenjamin

src/Core/Settings.h

@@ -1055,6 +1055,7 @@ class IColumn;
     M(Bool, input_format_json_ignore_unknown_keys_in_named_tuple, true, "Ignore unknown keys in json object for named tuples", 0) \
     M(Bool, input_format_json_defaults_for_missing_elements_in_named_tuple, true, "Insert default value in named tuple element if it's missing in json object", 0) \
     M(Bool, input_format_json_throw_on_bad_escape_sequence, true, "Throw an exception if JSON string contains bad escape sequence in JSON input formats. If disabled, bad escape sequences will remain as is in the data", 0) \
+    M(Bool, input_format_json_ignore_unnecessary_fields, true, "Ignore unnecessary fields and not parse them. Enabling this may not throw exceptions on json strings of invalid format or with duplicated fields", 0) \
     M(Bool, input_format_try_infer_integers, true, "Try to infer integers instead of floats while schema inference in text formats", 0) \
     M(Bool, input_format_try_infer_dates, true, "Try to infer dates from string fields while schema inference in text formats", 0) \
     M(Bool, input_format_try_infer_datetimes, true, "Try to infer datetimes from string fields while schema inference in text formats", 0) \

src/Core/SettingsChangesHistory.h

@@ -89,6 +89,7 @@ static std::map<ClickHouseVersion, SettingsChangesHistory::SettingsChanges> sett
               {"ignore_drop_queries_probability", 0, 0, "Allow to ignore drop queries in server with specified probability for testing purposes"},
               {"lightweight_deletes_sync", 2, 2, "The same as 'mutation_sync', but controls only execution of lightweight deletes"},
               {"query_cache_system_table_handling", "save", "throw", "The query cache no longer caches results of queries against system tables"},
+              {"input_format_json_ignore_unnecessary_fields", false, true, "Ignore unnecessary fields and not parse them. Enabling this may not throw exceptions on json strings of invalid format or with duplicated fields"},
               {"input_format_hive_text_allow_variable_number_of_columns", false, true, "Ignore extra columns in Hive Text input (if file has more columns than expected) and treat missing fields in Hive Text input as default values."},
               {"first_day_of_week", "Monday", "Monday", "Added a setting for the first day of the week for date/time functions"},
     }},

src/Formats/FormatFactory.cpp

@@ -144,6 +144,7 @@ FormatSettings getFormatSettings(const ContextPtr & context, const Settings & se
     format_settings.json.compact_allow_variable_number_of_columns = settings.input_format_json_compact_allow_variable_number_of_columns;
     format_settings.json.try_infer_objects_as_tuples = settings.input_format_json_try_infer_named_tuples_from_objects;
     format_settings.json.throw_on_bad_escape_sequence = settings.input_format_json_throw_on_bad_escape_sequence;
+    format_settings.json.ignore_unnecessary_fields = settings.input_format_json_ignore_unnecessary_fields;
     format_settings.null_as_default = settings.input_format_null_as_default;
     format_settings.decimal_trailing_zeros = settings.output_format_decimal_trailing_zeros;
     format_settings.parquet.row_group_rows = settings.output_format_parquet_row_group_size;

src/Formats/FormatSettings.h

@@ -223,6 +223,7 @@ struct FormatSettings
         bool try_infer_objects_as_tuples = false;
         bool infer_incomplete_types_as_strings = true;
         bool throw_on_bad_escape_sequence = true;
+        bool ignore_unnecessary_fields = true;
     } json{};
 
     struct

src/Processors/Formats/Impl/JSONEachRowRowInputFormat.cpp

@@ -132,6 +132,7 @@ void JSONEachRowRowInputFormat::readField(size_t index, MutableColumns & columns
         throw Exception(ErrorCodes::INCORRECT_DATA, "Duplicate field found while parsing JSONEachRow format: {}", columnName(index));
 
     seen_columns[index] = true;
+    seen_columns_count++;
     const auto & type = getPort().getHeader().getByPosition(index).type;
     const auto & serialization = serializations[index];
     read_columns[index] = JSONUtils::readField(*in, *columns[index], type, serialization, columnName(index), format_settings, yield_strings);
@@ -161,6 +162,14 @@ void JSONEachRowRowInputFormat::readJSONObject(MutableColumns & columns)
     for (size_t key_index = 0; advanceToNextKey(key_index); ++key_index)
     {
         StringRef name_ref = readColumnName(*in);
+        if (seen_columns_count >= total_columns && format_settings.json.ignore_unnecessary_fields)
+        {
+            // Keep parsing the remaining fields in case of the json is invalid.
+            // But not look up the name in the name_map since the cost cannot be ignored
+            JSONUtils::skipColon(*in);
+            skipUnknownField(name_ref);
+            continue;
+        }
         const size_t column_index = columnIndex(name_ref, key_index);
 
         if (unlikely(ssize_t(column_index) < 0))
@@ -210,6 +219,8 @@ bool JSONEachRowRowInputFormat::readRow(MutableColumns & columns, RowReadExtensi
         return false;
 
     size_t num_columns = columns.size();
+    total_columns = num_columns;
+    seen_columns_count = 0;
 
     read_columns.assign(num_columns, false);
     seen_columns.assign(num_columns, false);

src/Processors/Formats/Impl/JSONEachRowRowInputFormat.h

@@ -89,6 +89,8 @@ class JSONEachRowRowInputFormat : public IRowInputFormat
     std::vector<UInt8> read_columns;
     /// Set of columns which already met in row. Exception is thrown if there are more than one column with the same name.
     std::vector<UInt8> seen_columns;
+    size_t seen_columns_count = 0;
+    size_t total_columns = 0;
 
     /// This flag is needed to know if data is in square brackets.
     bool data_in_square_brackets = false;

tests/performance/json_input_format_part_fields.xml

@@ -0,0 +1,30 @@
+<test>
+    <create_query>
+    create table t(
+        f1 Nullable(String),
+        f2 Nullable(Int32),
+        f3 Nullable(String),
+        f4 Nullable(Int32),
+        f5 Nullable(UInt64),
+        f6 Nullable(UInt64),
+        f7 Nullable(String),
+        f8 Nullable(String)
+    ) ENGINE = File(JSONEachRow)</create_query>
+    <fill_query>
+    insert into t 
+    select 
+        '3014660362662815',
+        123,
+        'xxsdfsdfs',
+        22,
+        240321215532916041,
+        1711036533457,
+        'xxxxx',
+        'sdsfsdfsddf'
+    from numbers_mt(1000000)
+    </fill_query>
+
+    <query>select count(f1) from t settings max_threads=2</query>
+
+    <drop_query>drop table t</drop_query>
+</test>