|
49 | 49 | #include "paimon/testing/utils/read_result_collector.h" |
50 | 50 | #include "paimon/testing/utils/test_helper.h" |
51 | 51 | #include "paimon/testing/utils/testharness.h" |
| 52 | +#include "rapidjson/document.h" |
| 53 | +#include "rapidjson/stringbuffer.h" |
| 54 | +#include "rapidjson/writer.h" |
52 | 55 |
|
53 | 56 | namespace paimon::test { |
54 | 57 | // This is a sdk end-to-end test demo that supports write, commit, scan, and read operations. |
@@ -92,6 +95,38 @@ class WriteAndReadInteTest |
92 | 95 | return file_system->AtomicStore(schema_path, schema_content); |
93 | 96 | } |
94 | 97 |
|
| 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 | + |
95 | 130 | Result<bool> ReadAndCheckProjectedResult(const std::map<std::string, std::string>& options, |
96 | 131 | const std::vector<std::string>& read_fields, |
97 | 132 | const std::shared_ptr<arrow::DataType>& expected_type, |
@@ -930,6 +965,64 @@ TEST_P(WriteAndReadInteTest, TestWriteSamePartitionTwiceWithAllBasicTypesForPk) |
930 | 965 | ASSERT_TRUE(success); |
931 | 966 | } |
932 | 967 |
|
| 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"json([ |
| 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 | + ])json", |
| 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 | + |
933 | 1026 | std::vector<std::pair<std::string, std::string>> GetTestValuesForWriteAndReadInteTest() { |
934 | 1027 | std::vector<std::pair<std::string, std::string>> values = {{"parquet", "local"}}; |
935 | 1028 | // values.emplace_back("parquet", "jindo"); |
|
0 commit comments