feat(types): support CHAR/VARCHAR/BINARY/VARBINARY in data type json parser

SteNicholas · SteNicholas · commit 7e4484475e5f · 2026-06-30T19:20:04.000+08:00
Previously CHAR/VARCHAR/BINARY/VARBINARY were registered as keywords but had
no handling branch in ParseTypeByKeyword, so deserializing a schema that
contained them failed with "Unsupported type: VARCHAR". Map CHAR/VARCHAR to
arrow::utf8() and BINARY/VARBINARY to arrow::binary() (consistent with STRING
and BYTES), parsing the optional length parameter and validating it is within
[1, INT_MAX] (both inclusive), consistent with Java Paimon.

Add parser test cases covering these types with and without a length argument
(including the min/max length boundaries and invalid lengths), an end-to-end
write/read integration test exercising the types through the full Paimon flow,
and update the data type mapping doc accordingly.

Use a custom raw-string delimiter (R"json(...)json") for the schema JSON in the
integration test so that the embedded type strings such as "CHAR(10)" do not
terminate the raw string early at their internal )" sequence.
diff --git a/docs/source/user_guide/data_types.rst b/docs/source/user_guide/data_types.rst
@@ -38,48 +38,58 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
    * - ``CHAR``
 
        ``CHAR(n)``
-     - Not Supported
+     - Utf8
      - Data type of a fixed-length character string.
 
        The type can be declared using ``CHAR(n)`` where n is the number of code
        points. n must have a value between 1 and 2,147,483,647 (both inclusive).
        If no length is specified, n is equal to 1.
 
+       **Note:** Apache Arrow has no fixed-length string type, so the declared
+       length ``n`` is not enforced; values are mapped to a variable-length ``Utf8``.
+
    * - ``VARCHAR``
 
        ``VARCHAR(n)``
 
-     - Not Supported
+     - Utf8
      - Data type of a variable-length character string.
 
        The type can be declared using ``VARCHAR(n)`` where n is the maximum
        number of code points. n must have a value between 1 and 2,147,483,647
        (both inclusive). If no length is specified, n is equal to 1.
 
+       **Note:** The declared maximum length ``n`` is not enforced.
+
    * - ``STRING``
      - Utf8
      - Data type of a variable-length character string. ``STRING`` is a synonym for ``VARCHAR(2147483647)``.
 
    * - ``BINARY``
 
        ``BINARY(n)``
-     - Not Supported
+     - Binary
      - Data type of a fixed-length binary string (=a sequence of bytes).
 
        The type can be declared using ``BINARY(n)`` where n is the number of
        bytes. n must have a value between 1 and 2,147,483,647 (both inclusive).
        If no length is specified, n is equal to 1.
 
+       **Note:** Apache Arrow has no fixed-length binary type here, so the declared
+       length ``n`` is not enforced; values are mapped to a variable-length ``Binary``.
+
    * - ``VARBINARY``
 
        ``VARBINARY(n)``
-     - Not Supported
+     - Binary
      - Data type of a variable-length binary string (=a sequence of bytes).
 
        The type can be declared using ``VARBINARY(n)`` where n is the maximum
        number of bytes. n must have a value between 1 and 2,147,483,647
        (both inclusive). If no length is specified, n is equal to 1.
 
+       **Note:** The declared maximum length ``n`` is not enforced.
+
    * - ``BYTES``
      - Binary
      - ``BYTES`` is a synonym for ``VARBINARY(2147483647)``.
diff --git a/src/paimon/common/types/data_type_json_parser.cpp b/src/paimon/common/types/data_type_json_parser.cpp
@@ -20,6 +20,7 @@
 #include <cctype>
 #include <cstddef>
 #include <cstdint>
+#include <limits>
 #include <map>
 #include <sstream>
 #include <utility>
@@ -465,6 +466,12 @@ Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTypeWithNullability(b
 Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTypeByKeyword(bool* is_blob) {
     PAIMON_RETURN_NOT_OK(NextToken(TokenType::KEYWORD));
     switch (TokenAsKeyword()) {
+        case Keyword::CHAR:
+        case Keyword::VARCHAR:
+            return ParseStringType<arrow::StringType>();
+        case Keyword::BINARY:
+        case Keyword::VARBINARY:
+            return ParseStringType<arrow::BinaryType>();
         case Keyword::BYTES:
             return arrow::binary();
         case Keyword::BLOB: {
@@ -508,9 +515,14 @@ Result<int32_t> TokenParser::ParseStringLength() {
     if (HasNextToken({TokenType::BEGIN_PARAMETER})) {
         PAIMON_RETURN_NOT_OK(NextToken(TokenType::BEGIN_PARAMETER));
         PAIMON_RETURN_NOT_OK(NextToken(TokenType::LITERAL_INT));
-        auto length = TokenAsInt();
+        int64_t length = std::stoll(GetToken().value);
+        if (length < 1 || length > std::numeric_limits<int32_t>::max()) {
+            return Status::Invalid(
+                fmt::format("length must be between 1 and {} (both inclusive), but was {}",
+                            std::numeric_limits<int32_t>::max(), length));
+        }
         PAIMON_RETURN_NOT_OK(NextToken(TokenType::END_PARAMETER));
-        return length;
+        return static_cast<int32_t>(length);
     }
     // implicit length
     return -1;
diff --git a/src/paimon/common/types/data_type_json_parser_test.cpp b/src/paimon/common/types/data_type_json_parser_test.cpp
@@ -111,6 +111,16 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) {
         {"TIMESTAMP_LTZ(9)", arrow::timestamp(arrow::TimeUnit::NANO, timezone)},
         {"BYTES", arrow::binary()},
         {"STRING", arrow::utf8()},
+        {"CHAR", arrow::utf8()},
+        {"CHAR(10)", arrow::utf8()},
+        {"VARCHAR", arrow::utf8()},
+        {"VARCHAR(10)", arrow::utf8()},
+        {"BINARY", arrow::binary()},
+        {"BINARY(10)", arrow::binary()},
+        {"VARBINARY", arrow::binary()},
+        {"VARBINARY(10)", arrow::binary()},
+        {"CHAR(1)", arrow::utf8()},
+        {"VARCHAR(2147483647)", arrow::utf8()},
     };
 
     for (const auto& test_case : test_cases) {
@@ -131,6 +141,12 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) {
         rapidjson::Value value("VARCHAR(test)", invalid_doc.GetAllocator());
         ASSERT_NOK(DataTypeJsonParser::ParseType("field_name", value));
     }
+    for (const char* invalid_type : {"VARCHAR(0)", "VARBINARY(0)", "VARCHAR(2147483648)"}) {
+        rapidjson::Document invalid_doc;
+        rapidjson::Value value(invalid_type, invalid_doc.GetAllocator());
+        ASSERT_NOK_WITH_MSG(DataTypeJsonParser::ParseType("field_name", value),
+                            "length must be between 1 and 2147483647");
+    }
     {
         rapidjson::Document invalid_doc;
         rapidjson::Value value("TIMESTAMP(4)", invalid_doc.GetAllocator());
diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp
@@ -49,6 +49,9 @@
 #include "paimon/testing/utils/read_result_collector.h"
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
+#include "rapidjson/document.h"
+#include "rapidjson/stringbuffer.h"
+#include "rapidjson/writer.h"
 
 namespace paimon::test {
 // This is a sdk end-to-end test demo that supports write, commit, scan, and read operations.
@@ -92,6 +95,38 @@ class WriteAndReadInteTest
         return file_system->AtomicStore(schema_path, schema_content);
     }
 
+    Status WriteNextSchemaWithRawFieldTypes(const std::string& fields_json,
+                                            int32_t highest_field_id) const {
+        auto file_system = dir_->GetFileSystem();
+        std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+        SchemaManager schema_manager(file_system, table_path);
+        PAIMON_ASSIGN_OR_RAISE(auto latest_schema_opt, schema_manager.Latest());
+        if (!latest_schema_opt) {
+            return Status::Invalid("table schema does not exist");
+        }
+        auto next_schema = std::make_shared<TableSchema>(*latest_schema_opt.value());
+        next_schema->id_ = latest_schema_opt.value()->Id() + 1;
+        next_schema->highest_field_id_ = highest_field_id;
+        PAIMON_ASSIGN_OR_RAISE(std::string schema_content, next_schema->ToJsonString());
+
+        rapidjson::Document schema_doc;
+        schema_doc.Parse(schema_content.c_str());
+        rapidjson::Document fields_doc;
+        fields_doc.Parse(fields_json.c_str());
+        if (schema_doc.HasParseError() || fields_doc.HasParseError() ||
+            !schema_doc.HasMember("fields")) {
+            return Status::Invalid("failed to assemble schema json with raw field types");
+        }
+        schema_doc["fields"].CopyFrom(fields_doc, schema_doc.GetAllocator());
+        rapidjson::StringBuffer buffer;
+        rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+        schema_doc.Accept(writer);
+
+        std::string schema_path = PathUtil::JoinPath(schema_manager.SchemaDirectory(),
+                                                     "schema-" + std::to_string(next_schema->Id()));
+        return file_system->AtomicStore(schema_path, std::string(buffer.GetString()));
+    }
+
     Result<bool> ReadAndCheckProjectedResult(const std::map<std::string, std::string>& options,
                                              const std::vector<std::string>& read_fields,
                                              const std::shared_ptr<arrow::DataType>& expected_type,
@@ -930,6 +965,64 @@ TEST_P(WriteAndReadInteTest, TestWriteSamePartitionTwiceWithAllBasicTypesForPk)
     ASSERT_TRUE(success);
 }
 
+TEST_P(WriteAndReadInteTest, TestCharVarcharBinaryVarbinaryTypes) {
+    auto [file_format, file_system] = GetParam();
+    if (file_format != "parquet" && file_format != "orc") {
+        return;
+    }
+    arrow::FieldVector fields = {
+        arrow::field("c", arrow::utf8()), arrow::field("vc", arrow::utf8()),
+        arrow::field("b", arrow::binary()), arrow::field("vb", arrow::binary()),
+    };
+    std::map<std::string, std::string> options = {
+        {Options::MANIFEST_FORMAT, "avro"},  {Options::FILE_FORMAT, file_format},
+        {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "-1"},
+        {Options::FILE_SYSTEM, file_system},
+    };
+    if (file_system == "jindo") {
+        options = AddOptionsForJindo(options);
+    }
+    ASSERT_OK_AND_ASSIGN(
+        auto helper, TestHelper::Create(test_dir_, arrow::schema(fields), /*partition_keys=*/{},
+                                        /*primary_keys=*/{}, options, /*is_streaming_mode=*/false));
+
+    ASSERT_OK(WriteNextSchemaWithRawFieldTypes(R"json([
+        { "id" : 0, "name" : "c", "type" : "CHAR(10)" },
+        { "id" : 1, "name" : "vc", "type" : "VARCHAR(20)" },
+        { "id" : 2, "name" : "b", "type" : "BINARY(10)" },
+        { "id" : 3, "name" : "vb", "type" : "VARBINARY(20)" }
+    ])json",
+                                               /*highest_field_id=*/3));
+    helper.reset();
+    ASSERT_OK_AND_ASSIGN(helper,
+                         TestHelper::Create(PathUtil::JoinPath(test_dir_, "foo.db/bar"), options,
+                                            /*is_streaming_mode=*/false));
+
+    std::string data = R"([
+        ["alice", "hello world", "abc", "xyz123"],
+        [null, null, null, null]
+    ])";
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+                         TestHelper::MakeRecordBatch(arrow::struct_(fields), data,
+                                                     /*partition_map=*/{}, /*bucket=*/0, {}));
+    ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
+                                     /*expected_commit_messages=*/std::nullopt));
+
+    arrow::FieldVector fields_with_row_kind = fields;
+    fields_with_row_kind.insert(fields_with_row_kind.begin(),
+                                arrow::field("_VALUE_KIND", arrow::int8()));
+    auto data_type = arrow::struct_(fields_with_row_kind);
+    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
+                         helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
+    std::string expected_data = R"([
+        [0, "alice", "hello world", "abc", "xyz123"],
+        [0, null, null, null, null]
+    ])";
+    ASSERT_OK_AND_ASSIGN(bool success,
+                         helper->ReadAndCheckResult(data_type, data_splits, expected_data));
+    ASSERT_TRUE(success);
+}
+
 std::vector<std::pair<std::string, std::string>> GetTestValuesForWriteAndReadInteTest() {
     std::vector<std::pair<std::string, std::string>> values = {{"parquet", "local"}};
     // values.emplace_back("parquet", "jindo");
