Improve the error message when format schema isn't set.

dbms/src/Formats/CapnProtoRowInputStream.cpp

@@ -298,7 +298,7 @@ void registerInputFormatCapnProto(FormatFactory & factory)
         [](ReadBuffer & buf, const Block & sample, const Context & context, UInt64 max_block_size, const FormatSettings & settings)
         {
             return std::make_shared<BlockInputStreamFromRowInputStream>(
-                std::make_shared<CapnProtoRowInputStream>(buf, sample, FormatSchemaInfo(context, "capnp")),
+                std::make_shared<CapnProtoRowInputStream>(buf, sample, FormatSchemaInfo(context, "CapnProto")),
                 sample,
                 max_block_size,
                 settings);

dbms/src/Formats/FormatSchemaInfo.cpp

@@ -11,20 +11,29 @@ namespace ErrorCodes
     extern const int BAD_ARGUMENTS;
 }
 
-FormatSchemaInfo::FormatSchemaInfo(const Context & context, const String & schema_file_extension, bool schema_required)
+
+namespace
 {
-    String format_schema = context.getSettingsRef().format_schema.toString();
-    if (format_schema.empty())
+    String getFormatSchemaDefaultFileExtension(const String & format)
     {
-        if (schema_required)
-        {
-            throw Exception(
-                "Format schema requires the 'format_schema' setting to have the 'schema_file:message_name' format"
-                    + (schema_file_extension.empty() ? "" : ", e.g. 'schema." + schema_file_extension + ":Message'"),
-                ErrorCodes::BAD_ARGUMENTS);
-        }
-        return;
+        if (format == "Protobuf")
+            return "proto";
+        else if (format == "CapnProto")
+            return "capnp";
+        else
+            return "";
     }
+}
+
+
+FormatSchemaInfo::FormatSchemaInfo(const Context & context, const String & format)
+{
+    String format_schema = context.getSettingsRef().format_schema.toString();
+    if (format_schema.empty())
+        throw Exception(
+            "The format " + format + " requires a schema. The 'format_schema' setting should be set", ErrorCodes::BAD_ARGUMENTS);
+
+    String default_file_extension = getFormatSchemaDefaultFileExtension(format);
 
     size_t colon_pos = format_schema.find(':');
     Poco::Path path;
@@ -33,12 +42,11 @@ FormatSchemaInfo::FormatSchemaInfo(const Context & context, const String & schem
     {
         throw Exception(
             "Format schema requires the 'format_schema' setting to have the 'schema_file:message_name' format"
-                + (schema_file_extension.empty() ? "" : ", e.g. 'schema." + schema_file_extension + ":Message'") + ". Got '" + format_schema
+                + (default_file_extension.empty() ? "" : ", e.g. 'schema." + default_file_extension + ":Message'") + ". Got '" + format_schema
                 + "'",
             ErrorCodes::BAD_ARGUMENTS);
     }
 
-    is_null = false;
     message_name = format_schema.substr(colon_pos + 1);
 
     auto default_schema_directory = [&context]()
@@ -51,8 +59,8 @@ FormatSchemaInfo::FormatSchemaInfo(const Context & context, const String & schem
         return context.hasGlobalContext() && (context.getGlobalContext().getApplicationType() == Context::ApplicationType::SERVER);
     };
 
-    if (path.getExtension().empty() && !schema_file_extension.empty())
-        path.setExtension(schema_file_extension);
+    if (path.getExtension().empty() && !default_file_extension.empty())
+        path.setExtension(default_file_extension);
 
     if (path.isAbsolute())
     {

dbms/src/Formats/FormatSchemaInfo.h

@@ -10,10 +10,7 @@ class Context;
 class FormatSchemaInfo
 {
 public:
-    FormatSchemaInfo() = default;
-    FormatSchemaInfo(const Context & context, const String & schema_file_extension = String(), bool schema_required = true);
-
-    bool isNull() const { return is_null; }
+    FormatSchemaInfo(const Context & context, const String & format);
 
     /// Returns path to the schema file.
     const String & schemaPath() const { return schema_path; }
@@ -26,7 +23,6 @@ public:
     const String & messageName() const { return message_name; }
 
 private:
-    bool is_null = true;
     String schema_path;
     String schema_directory;
     String message_name;

dbms/src/Formats/ProtobufRowInputStream.cpp

@@ -75,7 +75,7 @@ void registerInputFormatProtobuf(FormatFactory & factory)
         const FormatSettings & settings)
     {
         return std::make_shared<BlockInputStreamFromRowInputStream>(
-            std::make_shared<ProtobufRowInputStream>(buf, sample, FormatSchemaInfo(context, "proto")),
+            std::make_shared<ProtobufRowInputStream>(buf, sample, FormatSchemaInfo(context, "Protobuf")),
             sample, max_block_size, settings);
     });
 }

dbms/src/Formats/ProtobufRowOutputStream.cpp

@@ -38,7 +38,7 @@ void registerOutputFormatProtobuf(FormatFactory & factory)
         "Protobuf", [](WriteBuffer & buf, const Block & header, const Context & context, const FormatSettings &)
         {
             return std::make_shared<BlockOutputStreamFromRowOutputStream>(
-                std::make_shared<ProtobufRowOutputStream>(buf, header, FormatSchemaInfo(context, "proto")), header);
+                std::make_shared<ProtobufRowOutputStream>(buf, header, FormatSchemaInfo(context, "Protobuf")), header);
         });
 }