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

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.

Author: SteNicholas

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)``.

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;

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());

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"([
+        { "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)" }
+    ])",
+                                               /*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");