fix(orc): support CHAR/VARCHAR compatibility and extend format compatibility tests (#362)

Author: lxy-9602

src/paimon/format/orc/orc_file_batch_reader.cpp

@@ -47,6 +47,36 @@
 namespace paimon::orc {
 namespace {
 
+bool IsStringTypeCompatible(const ::orc::Type* src_type, const ::orc::Type* target_type) {
+    ::orc::TypeKind src_kind = src_type->getKind();
+    return (src_kind == ::orc::TypeKind::CHAR || src_kind == ::orc::TypeKind::VARCHAR) &&
+           target_type->getKind() == ::orc::TypeKind::STRING;
+}
+
+bool IsCompleteTypeCompatible(const ::orc::Type* src_type, const ::orc::Type* target_type) {
+    ::orc::TypeKind src_kind = src_type->getKind();
+    ::orc::TypeKind target_kind = target_type->getKind();
+    if (IsStringTypeCompatible(src_type, target_type)) {
+        return true;
+    }
+    if (src_kind != target_kind || src_type->getSubtypeCount() != target_type->getSubtypeCount()) {
+        return false;
+    }
+    if (src_type->getSubtypeCount() == 0) {
+        return src_type->toString() == target_type->toString();
+    }
+    for (uint64_t i = 0; i < src_type->getSubtypeCount(); i++) {
+        if (src_kind == ::orc::TypeKind::STRUCT &&
+            src_type->getFieldName(i) != target_type->getFieldName(i)) {
+            return false;
+        }
+        if (!IsCompleteTypeCompatible(src_type->getSubtype(i), target_type->getSubtype(i))) {
+            return false;
+        }
+    }
+    return true;
+}
+
 void CollectAllColumnIds(const ::orc::Type* type, std::vector<uint64_t>* column_ids) {
     column_ids->push_back(type->getColumnId());
     for (uint64_t i = 0; i < type->getSubtypeCount(); ++i) {
@@ -208,7 +238,7 @@ Status OrcFileBatchReader::CollectTargetColumnIds(const ::orc::Type* src_type,
                                                   std::vector<uint64_t>* target_column_ids) {
     auto src_kind = src_type->getKind();
     auto target_kind = target_type->getKind();
-    if (src_kind != target_kind) {
+    if (src_kind != target_kind && !IsStringTypeCompatible(src_type, target_type)) {
         return Status::Invalid(fmt::format("type kind mismatch: src {} vs target {}",
                                            src_type->toString(), target_type->toString()));
     }
@@ -235,7 +265,7 @@ Status OrcFileBatchReader::CollectTargetColumnIds(const ::orc::Type* src_type,
             // Partial field recall inside list/map types is unsupported, so the target must match
             // the complete source subtree. Include the container and every descendant because all
             // of their streams are recalled by the ORC reader.
-            if (src_type->toString() != target_type->toString()) {
+            if (!IsCompleteTypeCompatible(src_type, target_type)) {
                 return Status::Invalid(fmt::format("type mismatch: src {} vs target {}",
                                                    src_type->toString(), target_type->toString()));
             }

src/paimon/format/orc/orc_file_batch_reader_test.cpp

@@ -533,6 +533,20 @@ TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsPrimitiveMap) {
     ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2, 3}));
 }
 
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsCompatibleStringKinds) {
+    std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(
+        "struct<char_value:char(3),varchar_value:varchar(64),"
+        "nested:array<map<char(4),varchar(16)>>>");
+    std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(
+        "struct<char_value:string,varchar_value:string,nested:array<map<string,string>>>");
+    std::vector<uint64_t> target_column_ids;
+
+    ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+                                                         &target_column_ids));
+    // char_value(1), varchar_value(2), nested-list(3), map(4), key(5), value(6)
+    ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2, 3, 4, 5, 6}));
+}
+
 TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsDeeplyNestedMap) {
     std::string schema =
         "struct<attributes:map<string,array<struct<score:double,tags:array<string>>>>>";

test/inte/paimon_read_compat_inte_test.cpp

@@ -0,0 +1,17 @@
+Schema: id INT NOT NULL, f_array_blob ARRAY<BLOB>.
+
+Options:
+
+bucket = -1
+file.format = avro
+blob-field = f_array_blob
+data-evolution.enabled = true
+row-tracking.enabled = true
+
+The table is unpartitioned and has no primary key.
+
+Add: (1, ["array-blob-value", null, ""])
+Add: (2, null)
+Add: (3, [])
+
+Paimon C++ is expected to reject this table while ARRAY<BLOB> is unsupported.

test/test_data/avro/append_types_compatibility.db/java_array_blob_types/README.md

@@ -0,0 +1,27 @@
+{
+  "version" : 3,
+  "id" : 0,
+  "fields" : [ {
+    "id" : 0,
+    "name" : "id",
+    "type" : "INT NOT NULL"
+  }, {
+    "id" : 1,
+    "name" : "f_array_blob",
+    "type" : {
+      "type" : "ARRAY",
+      "element" : "BLOB"
+    }
+  } ],
+  "highestFieldId" : 1,
+  "partitionKeys" : [ ],
+  "primaryKeys" : [ ],
+  "options" : {
+    "bucket" : "-1",
+    "blob-field" : "f_array_blob",
+    "row-tracking.enabled" : "true",
+    "data-evolution.enabled" : "true",
+    "file.format" : "avro"
+  },
+  "timeMillis" : 1789549988349
+}