lxy-9602 commented on code in PR #194: URL: https://github.com/apache/paimon-cpp/pull/194#discussion_r3772717382
########## src/paimon/core/index/pk/primary_key_index_definition.h: ########## @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include <cstdint> +#include <map> +#include <string> +#include <utility> + +namespace paimon { +/// Resolved definition of one source-backed primary-key index. +class PrimaryKeyIndexDefinition { + public: + /// Built-in primary-key index families. + enum class Family { + VECTOR, + BTREE, + BITMAP, + FULL_TEXT, + }; + + PrimaryKeyIndexDefinition(std::string column, int32_t field_id, std::string index_type, + std::map<std::string, std::string> options, Family family) + : column_(std::move(column)), Review Comment: Could we move the `Family family` parameter before `options`? ########## src/paimon/core/index/pk/primary_key_index_definitions.cpp: ########## @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/index/pk/primary_key_index_definitions.h" + +#include <set> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/utils/object_utils.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/defs.h" +#include "rapidjson/document.h" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" + +namespace paimon { +namespace { +constexpr char kBTreeIndexType[] = "btree"; +constexpr char kBitmapIndexType[] = "bitmap"; +constexpr char kFullTextIndexType[] = "full-text"; +constexpr char kBTreeOptionFamily[] = "pk-btree"; +constexpr char kBitmapOptionFamily[] = "pk-bitmap"; +constexpr char kBTreeAlgorithmPrefix[] = "btree-index."; +constexpr char kBitmapAlgorithmPrefix[] = "bitmap-index."; +constexpr char kFieldScopedPrefix[] = "fields."; +constexpr char kRecordsPerRangeKey[] = "sorted-index.records-per-range"; + +std::vector<std::string> IndexColumns(const std::map<std::string, std::string>& options, + const char* option_key) { + auto iter = options.find(option_key); + if (iter == options.end()) { + return {}; + } + std::vector<std::string> columns = StringUtils::Split(iter->second, ",", false); + for (std::string& column : columns) { + StringUtils::Trim(&column); + } + return columns; +} + +Status ValidateNoDuplicates(const std::vector<std::string>& columns, const char* option_key) { + std::set<std::string> unique_columns; + for (const std::string& column : columns) { + if (!unique_columns.insert(column).second) { + return Status::Invalid( + fmt::format("{} contains duplicate column '{}'.", option_key, column)); + } + } + return Status::OK(); +} + +Status ValidateUniqueColumns(std::set<std::string>* indexed_columns, + const std::vector<std::string>& columns) { + for (const std::string& column : columns) { + if (!indexed_columns->insert(column).second) { + return Status::Invalid( + fmt::format("Column '{}' can own at most one primary-key index.", column)); + } + } + return Status::OK(); +} Review Comment: I’m a bit curious whether `ValidateNoDuplicates` and `ValidateUniqueColumns` could be refactored into a shared helper function, with different error reporting as needed. Also, could we move the output parameter to the end of the parameter list? ########## src/paimon/common/utils/java_modified_utf8.h: ########## @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include <string> +#include <string_view> + +#include "paimon/result.h" + +namespace paimon { +/// Converts between standard UTF-8 and the "modified UTF-8" used by Java's +/// `DataOutputStream#writeUTF` / `DataInputStream#readUTF`: +/// - U+0000 is encoded as the two-byte sequence 0xC0 0x80 instead of a single zero byte; +/// - supplementary code points (U+10000 and above) are encoded as a UTF-16 surrogate pair, +/// each surrogate written as an independent three-byte sequence (CESU-8), instead of the +/// four-byte standard UTF-8 form. +class JavaModifiedUtf8 { + public: + JavaModifiedUtf8() = delete; Review Comment: This seems to be for binary compatibility of the source data file name in `PrimaryKeyIndexSourceMeta`, but in Java, under what scenarios would the file name contain Chinese characters or other special characters? ########## src/paimon/core/index/pk/primary_key_index_definitions_test.cpp: ########## @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/index/pk/primary_key_index_definitions.h" + +#include <map> +#include <memory> +#include <string> +#include <vector> + +#include "arrow/api.h" +#include "gtest/gtest.h" +#include "paimon/common/types/data_field.h" +#include "paimon/core/index/pk/primary_key_index_definition.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/defs.h" +#include "paimon/result.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { +/// Builds a primary-key table schema with fields id BIGINT (pk), price DOUBLE, age INT, +/// status STRING and emb FLOAT, merging the given options over a fixed bucket option. +Result<std::unique_ptr<TableSchema>> MakeSchema(std::map<std::string, std::string> options) { + std::vector<DataField> fields = { + DataField(0, arrow::field("id", arrow::int64(), /*nullable=*/false)), + DataField(1, arrow::field("price", arrow::float64())), + DataField(2, arrow::field("age", arrow::int32())), + DataField(3, arrow::field("status", arrow::utf8())), + DataField(4, arrow::field("emb", arrow::float32()))}; + options.emplace(Options::BUCKET, "1"); + return TableSchema::Create(/*schema_id=*/0, DataField::ConvertDataFieldsToArrowSchema(fields), + /*partition_keys=*/{}, /*primary_keys=*/{"id"}, options); +} +} // namespace + +TEST(PrimaryKeyIndexDefinitionsTest, NoIndexOptionsYieldsEmptyDefinitions) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, MakeSchema({})); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + ASSERT_TRUE(definitions.Definitions().empty()); + ASSERT_TRUE(definitions.ScalarDefinitions().empty()); +} + +TEST(PrimaryKeyIndexDefinitionsTest, ResolvesBTreeDefinitions) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "price,age"}})); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + ASSERT_EQ(2, definitions.Definitions().size()); + const PrimaryKeyIndexDefinition& price = definitions.Definitions()[0]; + ASSERT_EQ("price", price.Column()); + ASSERT_EQ(1, price.FieldId()); + ASSERT_EQ("btree", price.IndexType()); + ASSERT_EQ(PrimaryKeyIndexDefinition::Family::BTREE, price.GetFamily()); + const PrimaryKeyIndexDefinition& age = definitions.Definitions()[1]; + ASSERT_EQ("age", age.Column()); + ASSERT_EQ(2, age.FieldId()); + ASSERT_EQ("btree", age.IndexType()); + ASSERT_EQ(PrimaryKeyIndexDefinition::Family::BTREE, age.GetFamily()); + ASSERT_EQ(2, definitions.ScalarDefinitions().size()); +} + +TEST(PrimaryKeyIndexDefinitionsTest, ResolvesBitmapDefinition) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BITMAP_INDEX_COLUMNS, "status"}})); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + ASSERT_EQ(1, definitions.Definitions().size()); + const PrimaryKeyIndexDefinition& status = definitions.Definitions()[0]; + ASSERT_EQ("status", status.Column()); + ASSERT_EQ(3, status.FieldId()); + ASSERT_EQ("bitmap", status.IndexType()); + ASSERT_EQ(PrimaryKeyIndexDefinition::Family::BITMAP, status.GetFamily()); + ASSERT_EQ(1, definitions.ScalarDefinitions().size()); +} + +TEST(PrimaryKeyIndexDefinitionsTest, IgnoresColumnAbsentFromSchema) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "not_in_schema"}})); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + ASSERT_TRUE(definitions.Definitions().empty()); +} + +TEST(PrimaryKeyIndexDefinitionsTest, CoercesScalarJsonOptionValuesLikeJava) { + // Java's parseJsonMap(..., String.class) accepts scalar JSON values and coerces them + // to text, so numeric or boolean values written by a Java engine must stay readable. + std::map<std::string, std::string> options = { + {Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"fields.price.pk-btree.index.options", + R"({"compression-level":3,"cache-enabled":true,"block-size":"64 kb"})"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, MakeSchema(options)); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + const std::map<std::string, std::string>& resolved = definitions.Definitions()[0].Options(); + ASSERT_EQ("3", resolved.at("btree-index.compression-level")); + ASSERT_EQ("true", resolved.at("btree-index.cache-enabled")); + ASSERT_EQ("64 kb", resolved.at("btree-index.block-size")); + + // Null and nested values are rejected like in Java. + std::map<std::string, std::string> null_options = { + {Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"fields.price.pk-btree.index.options", R"({"block-size":null})"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> null_schema, MakeSchema(null_options)); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*null_schema)); + std::map<std::string, std::string> nested_options = { + {Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"fields.price.pk-btree.index.options", R"({"block-size":{"v":"64 kb"}})"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> nested_schema, MakeSchema(nested_options)); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*nested_schema)); +} + +TEST(PrimaryKeyIndexDefinitionsTest, QualifiesFieldScopedJsonOptions) { + std::map<std::string, std::string> options = { + {Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"sorted-index.records-per-range", "4096"}, + {"fields.price.pk-btree.index.options", + R"({"block-size":"64 kb","btree-index.cache-size":"32 mb","fields.foo.x":"y"})"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, MakeSchema(options)); + ASSERT_OK_AND_ASSIGN(PrimaryKeyIndexDefinitions definitions, + PrimaryKeyIndexDefinitions::Create(*schema)); + ASSERT_EQ(1, definitions.Definitions().size()); + const std::map<std::string, std::string>& resolved = definitions.Definitions()[0].Options(); + // Unqualified keys are prefixed with the algorithm prefix, qualified keys are kept as-is. + ASSERT_EQ(1, resolved.count("btree-index.block-size")); + ASSERT_EQ("64 kb", resolved.at("btree-index.block-size")); + ASSERT_EQ(1, resolved.count("btree-index.cache-size")); + ASSERT_EQ("32 mb", resolved.at("btree-index.cache-size")); + ASSERT_EQ(1, resolved.count("fields.foo.x")); + ASSERT_EQ("y", resolved.at("fields.foo.x")); + // The per-range knob never leaks into the definition, other table options are retained. + ASSERT_EQ(0, resolved.count("sorted-index.records-per-range")); + ASSERT_EQ(1, resolved.count(Options::BUCKET)); + ASSERT_EQ("1", resolved.at(Options::BUCKET)); +} + +TEST(PrimaryKeyIndexDefinitionsTest, RejectsConflictingJsonOptionValue) { + ASSERT_OK_AND_ASSIGN( + std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"btree-index.block-size", "128 kb"}, + {"fields.price.pk-btree.index.options", R"({"block-size":"64 kb"})"}})); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*schema)); +} + +TEST(PrimaryKeyIndexDefinitionsTest, RejectsMalformedJsonOptions) { + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"fields.price.pk-btree.index.options", "not-json"}})); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*schema)); + } + { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "price"}, + {"fields.price.pk-btree.index.options", R"({"":"v"})"}})); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*schema)); + } +} + +TEST(PrimaryKeyIndexDefinitionsTest, RejectsDuplicateColumnWithinFamily) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema, + MakeSchema({{Options::PK_BTREE_INDEX_COLUMNS, "price,price"}})); + ASSERT_NOK(PrimaryKeyIndexDefinitions::Create(*schema)); +} Review Comment: Please make the error message explicit. I’d recommend using `ASSERT_NOK_WITH_MSG`. ########## src/paimon/core/index/pk/primary_key_index_definitions.cpp: ########## @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/index/pk/primary_key_index_definitions.h" + +#include <set> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/utils/object_utils.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/defs.h" +#include "rapidjson/document.h" +#include "rapidjson/stringbuffer.h" +#include "rapidjson/writer.h" + +namespace paimon { +namespace { +constexpr char kBTreeIndexType[] = "btree"; +constexpr char kBitmapIndexType[] = "bitmap"; +constexpr char kFullTextIndexType[] = "full-text"; +constexpr char kBTreeOptionFamily[] = "pk-btree"; +constexpr char kBitmapOptionFamily[] = "pk-bitmap"; +constexpr char kBTreeAlgorithmPrefix[] = "btree-index."; +constexpr char kBitmapAlgorithmPrefix[] = "bitmap-index."; +constexpr char kFieldScopedPrefix[] = "fields."; +constexpr char kRecordsPerRangeKey[] = "sorted-index.records-per-range"; + +std::vector<std::string> IndexColumns(const std::map<std::string, std::string>& options, + const char* option_key) { + auto iter = options.find(option_key); + if (iter == options.end()) { + return {}; + } + std::vector<std::string> columns = StringUtils::Split(iter->second, ",", false); + for (std::string& column : columns) { + StringUtils::Trim(&column); + } + return columns; +} + +Status ValidateNoDuplicates(const std::vector<std::string>& columns, const char* option_key) { + std::set<std::string> unique_columns; + for (const std::string& column : columns) { + if (!unique_columns.insert(column).second) { + return Status::Invalid( + fmt::format("{} contains duplicate column '{}'.", option_key, column)); + } + } + return Status::OK(); +} + +Status ValidateUniqueColumns(std::set<std::string>* indexed_columns, + const std::vector<std::string>& columns) { + for (const std::string& column : columns) { + if (!indexed_columns->insert(column).second) { + return Status::Invalid( + fmt::format("Column '{}' can own at most one primary-key index.", column)); + } + } + return Status::OK(); +} + +/// Resolves the effective option map of one sorted-index definition: table options first, +/// then the field-scoped JSON options with unqualified keys prefixed by the algorithm +/// prefix, mirroring Java `CoreOptions#primaryKeySortedIndexOptions`. +Result<std::map<std::string, std::string>> SortedIndexOptions( + const std::map<std::string, std::string>& table_options, const std::string& column, + const char* option_family, const char* algorithm_prefix) { + std::map<std::string, std::string> resolved = table_options; + resolved.erase(kRecordsPerRangeKey); + std::string option_key = + fmt::format("{}{}.{}.index.options", kFieldScopedPrefix, column, option_family); + auto iter = table_options.find(option_key); + if (iter == table_options.end() || StringUtils::IsNullOrWhitespaceOnly(iter->second)) { + return resolved; + } + + rapidjson::Document document; + document.Parse(iter->second.c_str()); + if (document.HasParseError() || !document.IsObject()) { + return Status::Invalid( + fmt::format("{} must be a JSON object of option key-value pairs.", option_key)); + } + for (auto member = document.MemberBegin(); member != document.MemberEnd(); ++member) { + if (!member->name.IsString() || + StringUtils::IsNullOrWhitespaceOnly(member->name.GetString())) { + return Status::Invalid(fmt::format("{} contains an empty option key.", option_key)); + } + std::string key = member->name.GetString(); + if (member->value.IsNull()) { + return Status::Invalid( + fmt::format("{} value for key {} must not be null.", option_key, key)); + } + if (member->value.IsObject() || member->value.IsArray()) { + return Status::Invalid( + fmt::format("{} must be a JSON object of option key-value pairs.", option_key)); + } + std::string value; + if (member->value.IsString()) { + value = member->value.GetString(); + } else { + // Java's parseJsonMap(..., String.class) coerces scalar JSON values (numbers, + // booleans) to their text form, so `{"compression-level":3}` is valid there. + rapidjson::StringBuffer buffer; + rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); + member->value.Accept(writer); + value = buffer.GetString(); + } + std::string qualified_key = StringUtils::StartsWith(key, algorithm_prefix) || + StringUtils::StartsWith(key, kFieldScopedPrefix) + ? key + : algorithm_prefix + key; + auto previous = resolved.find(qualified_key); + if (previous != resolved.end() && previous->second != value) { + return Status::Invalid( + fmt::format("{} defines conflicting values for {}.", option_key, qualified_key)); + } + resolved[qualified_key] = value; + } + return resolved; +} + +} // namespace + +Result<PrimaryKeyIndexDefinitions> PrimaryKeyIndexDefinitions::Create(const TableSchema& schema) { + const std::map<std::string, std::string>& options = schema.Options(); + std::vector<std::string> vector_columns = + IndexColumns(options, Options::PK_VECTOR_INDEX_COLUMNS); + std::vector<std::string> btree_columns = IndexColumns(options, Options::PK_BTREE_INDEX_COLUMNS); + std::vector<std::string> bitmap_columns = + IndexColumns(options, Options::PK_BITMAP_INDEX_COLUMNS); + std::vector<std::string> full_text_columns = + IndexColumns(options, Options::PK_FULL_TEXT_INDEX_COLUMNS); + PAIMON_RETURN_NOT_OK(ValidateNoDuplicates(vector_columns, Options::PK_VECTOR_INDEX_COLUMNS)); + PAIMON_RETURN_NOT_OK(ValidateNoDuplicates(btree_columns, Options::PK_BTREE_INDEX_COLUMNS)); + PAIMON_RETURN_NOT_OK(ValidateNoDuplicates(bitmap_columns, Options::PK_BITMAP_INDEX_COLUMNS)); + PAIMON_RETURN_NOT_OK( + ValidateNoDuplicates(full_text_columns, Options::PK_FULL_TEXT_INDEX_COLUMNS)); + std::set<std::string> indexed_columns; + PAIMON_RETURN_NOT_OK(ValidateUniqueColumns(&indexed_columns, vector_columns)); + PAIMON_RETURN_NOT_OK(ValidateUniqueColumns(&indexed_columns, btree_columns)); + PAIMON_RETURN_NOT_OK(ValidateUniqueColumns(&indexed_columns, bitmap_columns)); + PAIMON_RETURN_NOT_OK(ValidateUniqueColumns(&indexed_columns, full_text_columns)); + + std::vector<PrimaryKeyIndexDefinition> definitions; + for (const DataField& field : schema.Fields()) { + const std::string& column = field.Name(); + if (ObjectUtils::Contains(btree_columns, column)) { + Result<std::map<std::string, std::string>> definition_options = + SortedIndexOptions(options, column, kBTreeOptionFamily, kBTreeAlgorithmPrefix); + PAIMON_RETURN_NOT_OK(definition_options.status()); Review Comment: Please use `ASSERT_OK_AND_ASSIGN` instead of calling `PAIMON_RETURN_NOT_OK` first and then accessing `value()`. If there are similar cases elsewhere, could you fix them as well? ########## src/paimon/common/global_index/btree/btree_global_indexer.cpp: ########## @@ -120,8 +123,9 @@ Result<std::shared_ptr<GlobalIndexReader>> BTreeGlobalIndexer::CreateReader( } read_buffer_size = static_cast<int32_t>(tmp_buffer_size); } - // TODO(lisizhuo.lsz): Allow users to specify an executor - std::shared_ptr<Executor> executor = CreateDefaultExecutor(); + // Readers are created per payload group and may coexist for many buckets. Share the + // process-wide executor instead of creating a dedicated thread pool for every group. + std::shared_ptr<Executor> executor = GetGlobalDefaultExecutor(); return std::make_shared<LazyFilteredBTreeReader>(read_buffer_size, files, key_type, file_reader, Review Comment: @lszskye Please evaluate whether using `GlobalDefaultExecutor` here could cause any issues. I’d lean toward passing the same executor to each reader instead. Also, the executor thread count in Java is based on `GLOBAL_INDEX_THREAD_NUM`, rather than using the machine core count as `GetGlobalDefaultExecutor()` does. A more complete approach would be to let `C++ GlobalIndexer:` accept an external executor and, like Java, have the scan layer create and pass it in based on `global-index.thread-num`, instead of having `BTreeGlobalIndexer` implicitly choose `GetGlobalDefaultExecutor()`. ########## src/paimon/common/utils/java_modified_utf8.h: ########## @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include <string> +#include <string_view> + +#include "paimon/result.h" + +namespace paimon { +/// Converts between standard UTF-8 and the "modified UTF-8" used by Java's +/// `DataOutputStream#writeUTF` / `DataInputStream#readUTF`: +/// - U+0000 is encoded as the two-byte sequence 0xC0 0x80 instead of a single zero byte; +/// - supplementary code points (U+10000 and above) are encoded as a UTF-16 surrogate pair, +/// each surrogate written as an independent three-byte sequence (CESU-8), instead of the +/// four-byte standard UTF-8 form. +class JavaModifiedUtf8 { + public: + JavaModifiedUtf8() = delete; Review Comment: I think I have a rough understanding of the issue now. It’s not limited to this spot—`FileIndexFormat`, `DataSplit.bucketPath`, and `DeletionFile.path` all have similar problems with Chinese characters when converting `string` to UTF. I’d suggest removing this part from the current PR for now, and then submitting a separate PR later to fully address all `writeUTF` / `readUTF` related issues. ########## src/paimon/core/index/pk/primary_key_index_source_meta.cpp: ########## @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "paimon/core/index/pk/primary_key_index_source_meta.h" + +#include <algorithm> +#include <limits> +#include <string> +#include <string_view> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/utils/java_modified_utf8.h" +#include "paimon/core/index/index_file_meta.h" + +namespace paimon { +namespace { +// Each serialized entry needs at least the two-byte writeUTF length and one int64 row count, +// mirroring the defensive source file count cap of the Java deserializer. +constexpr size_t kMinBytesPerSourceFile = sizeof(uint16_t) + sizeof(int64_t); +constexpr size_t kMaxInitialSourceFileCapacity = 1024; + +void AppendBigEndian32(int32_t value, std::string* out) { + auto bits = static_cast<uint32_t>(value); + out->push_back(static_cast<char>((bits >> 24) & 0xFF)); + out->push_back(static_cast<char>((bits >> 16) & 0xFF)); + out->push_back(static_cast<char>((bits >> 8) & 0xFF)); + out->push_back(static_cast<char>(bits & 0xFF)); +} + +void AppendBigEndian64(int64_t value, std::string* out) { + auto bits = static_cast<uint64_t>(value); + for (int32_t shift = 56; shift >= 0; shift -= 8) { + out->push_back(static_cast<char>((bits >> shift) & 0xFF)); + } +} + Review Comment: Please try to reuse `DataOutputStream`, `DataInputStream`, `MemorySegmentOutputStream` for endianness conversion and data input/output. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
