Skip to content

Commit 3dbddd7

Browse files
SteNicholasclaude
andcommitted
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. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018FbwEfDgqeq2EXXx8AXukd
1 parent 48ce42c commit 3dbddd7

4 files changed

Lines changed: 137 additions & 6 deletions

File tree

docs/source/user_guide/data_types.rst

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,48 +38,58 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
3838
* - ``CHAR``
3939

4040
``CHAR(n)``
41-
- Not Supported
41+
- Utf8
4242
- Data type of a fixed-length character string.
4343

4444
The type can be declared using ``CHAR(n)`` where n is the number of code
4545
points. n must have a value between 1 and 2,147,483,647 (both inclusive).
4646
If no length is specified, n is equal to 1.
4747

48+
**Note:** Apache Arrow has no fixed-length string type, so the declared
49+
length ``n`` is not enforced; values are mapped to a variable-length ``Utf8``.
50+
4851
* - ``VARCHAR``
4952

5053
``VARCHAR(n)``
5154

52-
- Not Supported
55+
- Utf8
5356
- Data type of a variable-length character string.
5457

5558
The type can be declared using ``VARCHAR(n)`` where n is the maximum
5659
number of code points. n must have a value between 1 and 2,147,483,647
5760
(both inclusive). If no length is specified, n is equal to 1.
5861

62+
**Note:** The declared maximum length ``n`` is not enforced.
63+
5964
* - ``STRING``
6065
- Utf8
6166
- Data type of a variable-length character string. ``STRING`` is a synonym for ``VARCHAR(2147483647)``.
6267

6368
* - ``BINARY``
6469

6570
``BINARY(n)``
66-
- Not Supported
71+
- Binary
6772
- Data type of a fixed-length binary string (=a sequence of bytes).
6873

6974
The type can be declared using ``BINARY(n)`` where n is the number of
7075
bytes. n must have a value between 1 and 2,147,483,647 (both inclusive).
7176
If no length is specified, n is equal to 1.
7277

78+
**Note:** Apache Arrow has no fixed-length binary type here, so the declared
79+
length ``n`` is not enforced; values are mapped to a variable-length ``Binary``.
80+
7381
* - ``VARBINARY``
7482

7583
``VARBINARY(n)``
76-
- Not Supported
84+
- Binary
7785
- Data type of a variable-length binary string (=a sequence of bytes).
7886

7987
The type can be declared using ``VARBINARY(n)`` where n is the maximum
8088
number of bytes. n must have a value between 1 and 2,147,483,647
8189
(both inclusive). If no length is specified, n is equal to 1.
8290

91+
**Note:** The declared maximum length ``n`` is not enforced.
92+
8393
* - ``BYTES``
8494
- Binary
8595
- ``BYTES`` is a synonym for ``VARBINARY(2147483647)``.

src/paimon/common/types/data_type_json_parser.cpp

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include <cctype>
2121
#include <cstddef>
2222
#include <cstdint>
23+
#include <limits>
2324
#include <map>
2425
#include <sstream>
2526
#include <utility>
@@ -465,6 +466,12 @@ Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTypeWithNullability(b
465466
Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTypeByKeyword(bool* is_blob) {
466467
PAIMON_RETURN_NOT_OK(NextToken(TokenType::KEYWORD));
467468
switch (TokenAsKeyword()) {
469+
case Keyword::CHAR:
470+
case Keyword::VARCHAR:
471+
return ParseStringType<arrow::StringType>();
472+
case Keyword::BINARY:
473+
case Keyword::VARBINARY:
474+
return ParseStringType<arrow::BinaryType>();
468475
case Keyword::BYTES:
469476
return arrow::binary();
470477
case Keyword::BLOB: {
@@ -508,9 +515,14 @@ Result<int32_t> TokenParser::ParseStringLength() {
508515
if (HasNextToken({TokenType::BEGIN_PARAMETER})) {
509516
PAIMON_RETURN_NOT_OK(NextToken(TokenType::BEGIN_PARAMETER));
510517
PAIMON_RETURN_NOT_OK(NextToken(TokenType::LITERAL_INT));
511-
auto length = TokenAsInt();
518+
int64_t length = std::stoll(GetToken().value);
519+
if (length < 1 || length > std::numeric_limits<int32_t>::max()) {
520+
return Status::Invalid(
521+
fmt::format("length must be between 1 and {} (both inclusive), but was {}",
522+
std::numeric_limits<int32_t>::max(), length));
523+
}
512524
PAIMON_RETURN_NOT_OK(NextToken(TokenType::END_PARAMETER));
513-
return length;
525+
return static_cast<int32_t>(length);
514526
}
515527
// implicit length
516528
return -1;

src/paimon/common/types/data_type_json_parser_test.cpp

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,16 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) {
111111
{"TIMESTAMP_LTZ(9)", arrow::timestamp(arrow::TimeUnit::NANO, timezone)},
112112
{"BYTES", arrow::binary()},
113113
{"STRING", arrow::utf8()},
114+
{"CHAR", arrow::utf8()},
115+
{"CHAR(10)", arrow::utf8()},
116+
{"VARCHAR", arrow::utf8()},
117+
{"VARCHAR(10)", arrow::utf8()},
118+
{"BINARY", arrow::binary()},
119+
{"BINARY(10)", arrow::binary()},
120+
{"VARBINARY", arrow::binary()},
121+
{"VARBINARY(10)", arrow::binary()},
122+
{"CHAR(1)", arrow::utf8()},
123+
{"VARCHAR(2147483647)", arrow::utf8()},
114124
};
115125

116126
for (const auto& test_case : test_cases) {
@@ -131,6 +141,12 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) {
131141
rapidjson::Value value("VARCHAR(test)", invalid_doc.GetAllocator());
132142
ASSERT_NOK(DataTypeJsonParser::ParseType("field_name", value));
133143
}
144+
for (const char* invalid_type : {"VARCHAR(0)", "VARBINARY(0)", "VARCHAR(2147483648)"}) {
145+
rapidjson::Document invalid_doc;
146+
rapidjson::Value value(invalid_type, invalid_doc.GetAllocator());
147+
ASSERT_NOK_WITH_MSG(DataTypeJsonParser::ParseType("field_name", value),
148+
"length must be between 1 and 2147483647");
149+
}
134150
{
135151
rapidjson::Document invalid_doc;
136152
rapidjson::Value value("TIMESTAMP(4)", invalid_doc.GetAllocator());

test/inte/write_and_read_inte_test.cpp

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@
4949
#include "paimon/testing/utils/read_result_collector.h"
5050
#include "paimon/testing/utils/test_helper.h"
5151
#include "paimon/testing/utils/testharness.h"
52+
#include "rapidjson/document.h"
53+
#include "rapidjson/stringbuffer.h"
54+
#include "rapidjson/writer.h"
5255

5356
namespace paimon::test {
5457
// This is a sdk end-to-end test demo that supports write, commit, scan, and read operations.
@@ -92,6 +95,38 @@ class WriteAndReadInteTest
9295
return file_system->AtomicStore(schema_path, schema_content);
9396
}
9497

98+
Status WriteNextSchemaWithRawFieldTypes(const std::string& fields_json,
99+
int32_t highest_field_id) const {
100+
auto file_system = dir_->GetFileSystem();
101+
std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
102+
SchemaManager schema_manager(file_system, table_path);
103+
PAIMON_ASSIGN_OR_RAISE(auto latest_schema_opt, schema_manager.Latest());
104+
if (!latest_schema_opt) {
105+
return Status::Invalid("table schema does not exist");
106+
}
107+
auto next_schema = std::make_shared<TableSchema>(*latest_schema_opt.value());
108+
next_schema->id_ = latest_schema_opt.value()->Id() + 1;
109+
next_schema->highest_field_id_ = highest_field_id;
110+
PAIMON_ASSIGN_OR_RAISE(std::string schema_content, next_schema->ToJsonString());
111+
112+
rapidjson::Document schema_doc;
113+
schema_doc.Parse(schema_content.c_str());
114+
rapidjson::Document fields_doc;
115+
fields_doc.Parse(fields_json.c_str());
116+
if (schema_doc.HasParseError() || fields_doc.HasParseError() ||
117+
!schema_doc.HasMember("fields")) {
118+
return Status::Invalid("failed to assemble schema json with raw field types");
119+
}
120+
schema_doc["fields"].CopyFrom(fields_doc, schema_doc.GetAllocator());
121+
rapidjson::StringBuffer buffer;
122+
rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
123+
schema_doc.Accept(writer);
124+
125+
std::string schema_path = PathUtil::JoinPath(schema_manager.SchemaDirectory(),
126+
"schema-" + std::to_string(next_schema->Id()));
127+
return file_system->AtomicStore(schema_path, std::string(buffer.GetString()));
128+
}
129+
95130
Result<bool> ReadAndCheckProjectedResult(const std::map<std::string, std::string>& options,
96131
const std::vector<std::string>& read_fields,
97132
const std::shared_ptr<arrow::DataType>& expected_type,
@@ -930,6 +965,64 @@ TEST_P(WriteAndReadInteTest, TestWriteSamePartitionTwiceWithAllBasicTypesForPk)
930965
ASSERT_TRUE(success);
931966
}
932967

968+
TEST_P(WriteAndReadInteTest, TestCharVarcharBinaryVarbinaryTypes) {
969+
auto [file_format, file_system] = GetParam();
970+
if (file_format != "parquet" && file_format != "orc") {
971+
return;
972+
}
973+
arrow::FieldVector fields = {
974+
arrow::field("c", arrow::utf8()), arrow::field("vc", arrow::utf8()),
975+
arrow::field("b", arrow::binary()), arrow::field("vb", arrow::binary()),
976+
};
977+
std::map<std::string, std::string> options = {
978+
{Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT, file_format},
979+
{Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "-1"},
980+
{Options::FILE_SYSTEM, file_system},
981+
};
982+
if (file_system == "jindo") {
983+
options = AddOptionsForJindo(options);
984+
}
985+
ASSERT_OK_AND_ASSIGN(
986+
auto helper, TestHelper::Create(test_dir_, arrow::schema(fields), /*partition_keys=*/{},
987+
/*primary_keys=*/{}, options, /*is_streaming_mode=*/false));
988+
989+
ASSERT_OK(WriteNextSchemaWithRawFieldTypes(R"([
990+
{ "id" : 0, "name" : "c", "type" : "CHAR(10)" },
991+
{ "id" : 1, "name" : "vc", "type" : "VARCHAR(20)" },
992+
{ "id" : 2, "name" : "b", "type" : "BINARY(10)" },
993+
{ "id" : 3, "name" : "vb", "type" : "VARBINARY(20)" }
994+
])",
995+
/*highest_field_id=*/3));
996+
helper.reset();
997+
ASSERT_OK_AND_ASSIGN(helper,
998+
TestHelper::Create(PathUtil::JoinPath(test_dir_, "foo.db/bar"), options,
999+
/*is_streaming_mode=*/false));
1000+
1001+
std::string data = R"([
1002+
["alice", "hello world", "abc", "xyz123"],
1003+
[null, null, null, null]
1004+
])";
1005+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
1006+
TestHelper::MakeRecordBatch(arrow::struct_(fields), data,
1007+
/*partition_map=*/{}, /*bucket=*/0, {}));
1008+
ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
1009+
/*expected_commit_messages=*/std::nullopt));
1010+
1011+
arrow::FieldVector fields_with_row_kind = fields;
1012+
fields_with_row_kind.insert(fields_with_row_kind.begin(),
1013+
arrow::field("_VALUE_KIND", arrow::int8()));
1014+
auto data_type = arrow::struct_(fields_with_row_kind);
1015+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
1016+
helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
1017+
std::string expected_data = R"([
1018+
[0, "alice", "hello world", "abc", "xyz123"],
1019+
[0, null, null, null, null]
1020+
])";
1021+
ASSERT_OK_AND_ASSIGN(bool success,
1022+
helper->ReadAndCheckResult(data_type, data_splits, expected_data));
1023+
ASSERT_TRUE(success);
1024+
}
1025+
9331026
std::vector<std::pair<std::string, std::string>> GetTestValuesForWriteAndReadInteTest() {
9341027
std::vector<std::pair<std::string, std::string>> values = {{"parquet", "local"}};
9351028
// values.emplace_back("parquet", "jindo");

0 commit comments

Comments
 (0)