make it thread-safe and simplify ctor (#3326)

lang/c++/impl/Compiler.cc

@@ -399,6 +399,12 @@ static LogicalType makeLogicalType(const Entity &e, const Object &m) {
         t = LogicalType::DURATION;
     else if (typeField == "uuid")
         t = LogicalType::UUID;
+    else {
+        auto custom = CustomLogicalTypeRegistry::instance().create(typeField, e.toString());
+        if (custom != nullptr) {
+            return LogicalType(std::move(custom));
+        }
+    }
     return LogicalType(t);
 }
 

lang/c++/impl/LogicalType.cc

@@ -22,7 +22,14 @@
 namespace avro {
 
 LogicalType::LogicalType(Type type)
-    : type_(type), precision_(0), scale_(0) {}
+    : type_(type), precision_(0), scale_(0), custom_(nullptr) {
+    if (type == CUSTOM) {
+        throw Exception("Logical type CUSTOM must be initialized with a custom logical type");
+    }
+}
+
+LogicalType::LogicalType(std::shared_ptr<CustomLogicalType> custom)
+    : type_(CUSTOM), precision_(0), scale_(0), custom_(std::move(custom)) {}
 
 LogicalType::Type LogicalType::type() const {
     return type_;
@@ -92,7 +99,33 @@ void LogicalType::printJson(std::ostream &os) const {
         case UUID:
             os << R"("logicalType": "uuid")";
             break;
+        case CUSTOM:
+            custom_->printJson(os);
+            break;
+    }
+}
+
+void CustomLogicalType::printJson(std::ostream &os) const {
+    os << R"("logicalType": ")" << name_ << "\"";
+}
+
+CustomLogicalTypeRegistry &CustomLogicalTypeRegistry::instance() {
+    static CustomLogicalTypeRegistry instance;
+    return instance;
+}
+
+void CustomLogicalTypeRegistry::registerType(const std::string &name, Factory factory) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    registry_[name] = factory;
+}
+
+std::shared_ptr<CustomLogicalType> CustomLogicalTypeRegistry::create(const std::string &name, const std::string &json) const {
+    std::lock_guard<std::mutex> lock(mutex_);
+    auto it = registry_.find(name);
+    if (it == registry_.end()) {
+        return nullptr;
     }
+    return it->second(json);
 }
 
 } // namespace avro

lang/c++/impl/Node.cc

@@ -233,6 +233,11 @@ void Node::setLogicalType(LogicalType logicalType) {
                                 "STRING type");
             }
             break;
+        case LogicalType::CUSTOM:
+            if (logicalType.customLogicalType() == nullptr) {
+                throw Exception("CUSTOM logical type is not set");
+            }
+            break;
     }
 
     logicalType_ = logicalType;

lang/c++/impl/NodeImpl.cc

@@ -252,9 +252,18 @@ static void printName(std::ostream &os, const Name &n, size_t depth) {
     os << indent(depth) << R"("name": ")" << n.simpleName() << "\",\n";
 }
 
+static void printLogicalType(std::ostream &os, const LogicalType &logicalType, size_t depth) {
+    if (logicalType.type() != LogicalType::NONE) {
+        os << indent(depth);
+        logicalType.printJson(os);
+        os << ",\n";
+    }
+}
+
 void NodeRecord::printJson(std::ostream &os, size_t depth) const {
     os << "{\n";
     os << indent(++depth) << "\"type\": \"record\",\n";
+    printLogicalType(os, logicalType(), depth);
     const Name &name = nameAttribute_.get();
     printName(os, name, depth);
 
@@ -524,6 +533,7 @@ void NodeMap::printDefaultToJson(const GenericDatum &g, std::ostream &os,
 void NodeEnum::printJson(std::ostream &os, size_t depth) const {
     os << "{\n";
     os << indent(++depth) << "\"type\": \"enum\",\n";
+    printLogicalType(os, logicalType(), depth);
     if (!getDoc().empty()) {
         os << indent(depth) << R"("doc": ")"
            << escape(getDoc()) << "\",\n";
@@ -550,6 +560,7 @@ void NodeEnum::printJson(std::ostream &os, size_t depth) const {
 void NodeArray::printJson(std::ostream &os, size_t depth) const {
     os << "{\n";
     os << indent(depth + 1) << "\"type\": \"array\",\n";
+    printLogicalType(os, logicalType(), depth + 1);
     if (!getDoc().empty()) {
         os << indent(depth + 1) << R"("doc": ")"
            << escape(getDoc()) << "\",\n";
@@ -566,6 +577,7 @@ void NodeArray::printJson(std::ostream &os, size_t depth) const {
 void NodeMap::printJson(std::ostream &os, size_t depth) const {
     os << "{\n";
     os << indent(depth + 1) << "\"type\": \"map\",\n";
+    printLogicalType(os, logicalType(), depth + 1);
     if (!getDoc().empty()) {
         os << indent(depth + 1) << R"("doc": ")"
            << escape(getDoc()) << "\",\n";

lang/c++/include/avro/LogicalType.hh

@@ -19,12 +19,18 @@
 #ifndef avro_LogicalType_hh__
 #define avro_LogicalType_hh__
 
+#include <functional>
 #include <iostream>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
 
 #include "Config.hh"
 
 namespace avro {
 
+class CustomLogicalType;
+
 class AVRO_DECL LogicalType {
 public:
     enum Type {
@@ -41,10 +47,12 @@ public:
         LOCAL_TIMESTAMP_MICROS,
         LOCAL_TIMESTAMP_NANOS,
         DURATION,
-        UUID
+        UUID,
+        CUSTOM // for registered custom logical types
     };
 
     explicit LogicalType(Type type);
+    explicit LogicalType(std::shared_ptr<CustomLogicalType> custom);
 
     Type type() const;
 
@@ -57,12 +65,53 @@ public:
     void setScale(int32_t scale);
     int32_t scale() const { return scale_; }
 
+    const std::shared_ptr<CustomLogicalType> &customLogicalType() const {
+        return custom_;
+    }
+
     void printJson(std::ostream &os) const;
 
 private:
     Type type_;
     int32_t precision_;
     int32_t scale_;
+    std::shared_ptr<CustomLogicalType> custom_;
+};
+
+class AVRO_DECL CustomLogicalType {
+public:
+    CustomLogicalType(const std::string &name) : name_(name) {}
+
+    virtual ~CustomLogicalType() = default;
+
+    const std::string &name() const { return name_; }
+
+    virtual void printJson(std::ostream &os) const;
+
+private:
+    std::string name_;
+};
+
+// Registry for custom logical types.
+// This class is thread-safe.
+class AVRO_DECL CustomLogicalTypeRegistry {
+public:
+    static CustomLogicalTypeRegistry &instance();
+
+    using Factory = std::function<std::shared_ptr<CustomLogicalType>(const std::string &json)>;
+
+    // Register a custom logical type and its factory function.
+    void registerType(const std::string &name, Factory factory);
+
+    // Create a custom logical type from a JSON string.
+    // Returns nullptr if the name is not registered.
+    std::shared_ptr<CustomLogicalType> create(const std::string &name, const std::string &json) const;
+
+private:
+    CustomLogicalTypeRegistry() = default;
+
+    std::unordered_map<std::string, Factory> registry_;
+    mutable std::mutex mutex_;
 };
 
 } // namespace avro

lang/c++/test/SchemaTests.cc

@@ -658,6 +658,43 @@ static void testMalformedLogicalTypes(const char *schema) {
     BOOST_CHECK(datum.logicalType().type() == LogicalType::NONE);
 }
 
+static void testCustomLogicalType() {
+    // Declare a custom logical type.
+    struct MapLogicalType : public CustomLogicalType {
+        MapLogicalType() : CustomLogicalType("map") {}
+    };
+
+    // Register the custom logical type with the registry.
+    CustomLogicalTypeRegistry::instance().registerType("map", [](const std::string &) {
+        return std::make_shared<MapLogicalType>();
+    });
+
+    auto verifyCustomLogicalType = [](const ValidSchema &schema) {
+        auto logicalType = schema.root()->logicalType();
+        BOOST_CHECK_EQUAL(logicalType.type(), LogicalType::CUSTOM);
+        BOOST_CHECK_EQUAL(logicalType.customLogicalType()->name(), "map");
+    };
+
+    const std::string schema =
+        R"({ "type": "array",
+             "logicalType": "map",
+             "items": {
+               "type": "record",
+               "name": "k12_v13",
+               "fields": [
+                 { "name": "key", "type": "int", "field-id": 12 },
+                 { "name": "value", "type": "string", "field-id": 13 }
+               ]
+             }
+           })";
+    auto compiledSchema = compileJsonSchemaFromString(schema);
+    verifyCustomLogicalType(compiledSchema);
+
+    auto json = compiledSchema.toJson();
+    auto parsedSchema = compileJsonSchemaFromString(json);
+    verifyCustomLogicalType(parsedSchema);
+}
+
 } // namespace schema
 } // namespace avro
 
@@ -681,5 +718,6 @@ init_unit_test_suite(int /*argc*/, char * /*argv*/[]) {
     ADD_PARAM_TEST(ts, avro::schema::testMalformedLogicalTypes,
                    avro::schema::malformedLogicalTypes);
     ts->add(BOOST_TEST_CASE(&avro::schema::testCompactSchemas));
+    ts->add(BOOST_TEST_CASE(&avro::schema::testCustomLogicalType));
     return ts;
 }