Skip to content

Commit 9752cdf

Browse files
dalingmengdalingmeng
committed
feat(shared-shredding): Expose shared-shredding utilities
1 parent ffbda4f commit 9752cdf

5 files changed

Lines changed: 280 additions & 34 deletions

File tree

‎src/paimon/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,7 @@ set(PAIMON_COMMON_SRCS
139139
common/utils/crc32c.cpp
140140
common/utils/decimal_utils.cpp
141141
common/data/shredding/map_shared_shredding_utils.cpp
142+
common/data/shredding/map_shared_shredding_schema_utils.cpp
142143
common/data/shredding/map_shared_shredding_context.cpp
143144
common/data/shredding/map_shared_shredding_batch_converter.cpp
144145
common/data/shredding/map_shared_shredding_column_allocator.cpp
@@ -547,6 +548,7 @@ if(PAIMON_BUILD_TESTS)
547548
common/utils/threadsafe_queue_test.cpp
548549
common/utils/generic_lru_cache_test.cpp
549550
common/data/shredding/map_shared_shredding_utils_test.cpp
551+
common/data/shredding/map_shared_shredding_schema_utils_test.cpp
550552
common/data/shredding/map_shared_shredding_batch_converter_test.cpp
551553
common/data/shredding/lru_map_shared_shredding_column_allocator_test.cpp
552554
common/data/shredding/plain_map_shared_shredding_column_allocator_test.cpp

‎src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/data/shredding/map_shared_shredding_schema_utils.h"
18+
19+
#include "arrow/c/bridge.h"
20+
#include "arrow/type.h"
21+
#include "arrow/util/key_value_metadata.h"
22+
#include "fmt/format.h"
23+
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
24+
#include "paimon/common/utils/arrow/status_utils.h"
25+
26+
namespace paimon {
27+
28+
Result<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema(
29+
std::unique_ptr<::ArrowSchema> logical_schema,
30+
const std::map<std::string, int32_t>& field_to_num_columns) {
31+
if (!logical_schema) {
32+
return Status::Invalid("logical schema is null");
33+
}
34+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_logical_schema,
35+
arrow::ImportSchema(logical_schema.get()));
36+
std::shared_ptr<arrow::Schema> physical_schema =
37+
MapSharedShreddingUtils::LogicalToPhysicalSchema(arrow_logical_schema,
38+
field_to_num_columns);
39+
auto c_schema = std::make_unique<::ArrowSchema>();
40+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*physical_schema, c_schema.get()));
41+
return c_schema;
42+
}
43+
44+
Result<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
45+
std::unique_ptr<::ArrowSchema> physical_schema,
46+
const std::map<std::string, MapSharedShreddingFieldMeta>& field_name_to_meta,
47+
const std::string& compression) {
48+
if (!physical_schema) {
49+
return Status::Invalid("physical schema is null");
50+
}
51+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_physical_schema,
52+
arrow::ImportSchema(physical_schema.get()));
53+
54+
arrow::FieldVector updated_fields = arrow_physical_schema->fields();
55+
for (const auto& [field_name, field_meta] : field_name_to_meta) {
56+
int32_t field_index = arrow_physical_schema->GetFieldIndex(field_name);
57+
if (field_index < 0) {
58+
return Status::Invalid(fmt::format(
59+
"Shared-shredding field '{}' not found in physical schema.", field_name));
60+
}
61+
62+
const auto& field = arrow_physical_schema->field(field_index);
63+
auto metadata = field->metadata() ? field->metadata()->Copy()
64+
: std::make_shared<arrow::KeyValueMetadata>();
65+
PAIMON_RETURN_NOT_OK(
66+
MapSharedShreddingUtils::SerializeMetadata(field_meta, compression, metadata.get()));
67+
updated_fields[field_index] = field->WithMetadata(metadata);
68+
}
69+
70+
auto updated_schema =
71+
arrow::schema(std::move(updated_fields), arrow_physical_schema->metadata());
72+
auto c_schema = std::make_unique<::ArrowSchema>();
73+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*updated_schema, c_schema.get()));
74+
return c_schema;
75+
}
76+
77+
Result<MapSharedShreddingFieldMeta> MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
78+
std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name,
79+
const std::string& compression) {
80+
if (!physical_schema) {
81+
return Status::Invalid("physical schema is null");
82+
}
83+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_physical_schema,
84+
arrow::ImportSchema(physical_schema.get()));
85+
const auto& field = arrow_physical_schema->GetFieldByName(field_name);
86+
if (!field) {
87+
return Status::Invalid(
88+
fmt::format("Shared-shredding field '{}' not found in physical schema.", field_name));
89+
}
90+
91+
auto metadata =
92+
field->metadata() ? field->metadata()->Copy() : std::shared_ptr<arrow::KeyValueMetadata>();
93+
return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression);
94+
}
95+
96+
} // namespace paimon

‎src/paimon/common/data/shredding/map_shared_shredding_schema_utils_test.cpp‎

Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,171 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/data/shredding/map_shared_shredding_schema_utils.h"
18+
19+
#include <string>
20+
#include <utility>
21+
22+
#include "arrow/c/bridge.h"
23+
#include "arrow/type.h"
24+
#include "arrow/util/key_value_metadata.h"
25+
#include "gtest/gtest.h"
26+
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
27+
#include "paimon/testing/utils/testharness.h"
28+
29+
namespace paimon::test {
30+
31+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaBasic) {
32+
MapSharedShreddingFieldMeta tags_meta;
33+
tags_meta.name_to_id = {{"host", 0}, {"region", 1}};
34+
tags_meta.field_to_columns = {{0, {0}}, {1, {1}}};
35+
tags_meta.num_columns = 2;
36+
tags_meta.max_row_width = 2;
37+
38+
auto id_metadata = std::make_shared<arrow::KeyValueMetadata>();
39+
id_metadata->Append("paimon.field.id", "1");
40+
auto schema_metadata = std::make_shared<arrow::KeyValueMetadata>();
41+
schema_metadata->Append("schema.key", "schema.value");
42+
auto schema = arrow::schema(
43+
{arrow::field("id", arrow::int32(), true, id_metadata),
44+
arrow::field("tags", arrow::struct_({arrow::field("__field_mapping",
45+
arrow::list(arrow::int32()), true),
46+
arrow::field("__col_0", arrow::utf8(), true),
47+
arrow::field("__col_1", arrow::utf8(), true)}))},
48+
schema_metadata);
49+
50+
auto c_schema = std::make_unique<::ArrowSchema>();
51+
EXPECT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
52+
ASSERT_OK_AND_ASSIGN(auto c_updated_schema,
53+
MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
54+
std::move(c_schema), {{"tags", tags_meta}}, "none"));
55+
ASSERT_TRUE(c_updated_schema);
56+
ASSERT_TRUE(c_updated_schema->release);
57+
auto updated_schema = arrow::ImportSchema(c_updated_schema.get()).ValueOrDie();
58+
59+
ASSERT_TRUE(updated_schema->metadata()->Equals(*schema_metadata));
60+
ASSERT_TRUE(updated_schema->field(0)->metadata()->Equals(*id_metadata));
61+
62+
auto tags_metadata = updated_schema->GetFieldByName("tags")->metadata()->Copy();
63+
ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(tags_metadata));
64+
ASSERT_OK_AND_ASSIGN(auto deserialized,
65+
MapSharedShreddingUtils::DeserializeMetadata(tags_metadata, "none"));
66+
ASSERT_EQ(deserialized, tags_meta);
67+
}
68+
69+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaMissingField) {
70+
MapSharedShreddingFieldMeta tags_meta;
71+
tags_meta.name_to_id = {{"host", 0}};
72+
tags_meta.field_to_columns = {{0, {0}}};
73+
tags_meta.num_columns = 1;
74+
tags_meta.max_row_width = 1;
75+
76+
auto schema = arrow::schema({arrow::field("id", arrow::int32())});
77+
78+
auto c_schema = std::make_unique<::ArrowSchema>();
79+
EXPECT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
80+
ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
81+
std::move(c_schema), {{"tags", tags_meta}}, "none"),
82+
"Shared-shredding field 'tags' not found in physical schema.");
83+
}
84+
85+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaPreservesExistingFieldMetadata) {
86+
MapSharedShreddingFieldMeta tags_meta;
87+
tags_meta.name_to_id = {{"host", 0}};
88+
tags_meta.field_to_columns = {{0, {0}}};
89+
tags_meta.num_columns = 1;
90+
tags_meta.max_row_width = 1;
91+
92+
auto tags_metadata = std::make_shared<arrow::KeyValueMetadata>();
93+
tags_metadata->Append("paimon.field.id", "7");
94+
tags_metadata->Append("description", "original tags field");
95+
auto schema = arrow::schema({arrow::field(
96+
"tags",
97+
arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
98+
arrow::field("__col_0", arrow::utf8(), true)}),
99+
true, tags_metadata)});
100+
101+
auto c_schema = std::make_unique<::ArrowSchema>();
102+
EXPECT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
103+
ASSERT_OK_AND_ASSIGN(auto c_updated_schema,
104+
MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
105+
std::move(c_schema), {{"tags", tags_meta}}, "none"));
106+
ASSERT_TRUE(c_updated_schema);
107+
ASSERT_TRUE(c_updated_schema->release);
108+
auto updated_schema = arrow::ImportSchema(c_updated_schema.get()).ValueOrDie();
109+
110+
auto updated_metadata = updated_schema->field(0)->metadata()->Copy();
111+
ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("paimon.field.id")), "7");
112+
ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("description")),
113+
"original tags field");
114+
ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(updated_metadata));
115+
}
116+
117+
TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromField) {
118+
MapSharedShreddingFieldMeta tags_meta;
119+
tags_meta.name_to_id = {{"host", 0}, {"region", 1}};
120+
tags_meta.field_to_columns = {{0, {0}}, {1, {1}}};
121+
tags_meta.num_columns = 2;
122+
tags_meta.max_row_width = 2;
123+
124+
auto metadata = std::make_shared<arrow::KeyValueMetadata>();
125+
ASSERT_OK(MapSharedShreddingUtils::SerializeMetadata(tags_meta, "none", metadata.get()));
126+
auto field = arrow::field(
127+
"tags",
128+
arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
129+
arrow::field("__col_0", arrow::utf8(), true),
130+
arrow::field("__col_1", arrow::utf8(), true)}),
131+
true, metadata);
132+
auto schema = arrow::schema({arrow::field("id", arrow::int32()), field});
133+
134+
auto c_schema = std::make_unique<::ArrowSchema>();
135+
EXPECT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
136+
ASSERT_OK_AND_ASSIGN(auto parsed_meta, MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
137+
std::move(c_schema), "tags", "none"));
138+
ASSERT_EQ(parsed_meta, tags_meta);
139+
}
140+
141+
TEST(MapSharedShreddingSchemaUtilsTest, LogicalToPhysicalSchemaNestedListValue) {
142+
// MAP<STRING, STRUCT<a:array<int32>, b: array<int32>, c: array<int32>>>
143+
auto nested_value = arrow::struct_({arrow::field("a", arrow::list(arrow::int32())),
144+
arrow::field("b", arrow::list(arrow::int32())),
145+
arrow::field("c", arrow::list(arrow::int32()))});
146+
auto map_type = arrow::map(arrow::utf8(), nested_value);
147+
auto schema =
148+
arrow::schema({arrow::field("ts", arrow::int64()), arrow::field("data", map_type)});
149+
150+
std::map<std::string, int32_t> field_to_num_columns = {{"data", 2}};
151+
auto c_schema = std::make_unique<::ArrowSchema>();
152+
EXPECT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
153+
ASSERT_OK_AND_ASSIGN(auto c_physical_schema,
154+
MapSharedShreddingSchemaUtils::LogicalToPhysicalSchema(
155+
std::move(c_schema), field_to_num_columns));
156+
ASSERT_TRUE(c_physical_schema);
157+
ASSERT_TRUE(c_physical_schema->release);
158+
auto physical_schema = arrow::ImportSchema(c_physical_schema.get()).ValueOrDie();
159+
160+
auto expected_struct = arrow::struct_({
161+
arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
162+
arrow::field("__col_0", nested_value, true),
163+
arrow::field("__col_1", nested_value, true),
164+
arrow::field("__overflow", arrow::map(arrow::int32(), nested_value), true),
165+
});
166+
auto expected_schema = arrow::schema(
167+
{arrow::field("ts", arrow::int64()), arrow::field("data", expected_struct, true)});
168+
ASSERT_TRUE(physical_schema->Equals(expected_schema));
169+
}
170+
171+
} // namespace paimon::test

‎src/paimon/common/data/shredding/map_shared_shredding_utils.h‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626

2727
#include "arrow/type.h"
2828
#include "paimon/common/data/shredding/map_shredding_defs.h"
29+
#include "paimon/data/shredding/map_shared_shredding_schema_utils.h"
2930
#include "paimon/result.h"
3031
#include "paimon/status.h"
3132

@@ -83,6 +84,14 @@ class MapSharedShreddingUtils {
8384

8485
// ---- Metadata serialization ----
8586

87+
/// Serializes shredding metadata and appends entries to an existing KeyValueMetadata.
88+
/// @param field_meta The field-level shredding metadata to serialize.
89+
/// @param compression Compression codec name for field_dict compression.
90+
/// @param[out] metadata The KeyValueMetadata to append entries to.
91+
static Status SerializeMetadata(const MapSharedShreddingFieldMeta& field_meta,
92+
const std::string& compression,
93+
arrow::KeyValueMetadata* metadata);
94+
8695
/// Deserializes shredding metadata from file footer KeyValueMetadata (per field).
8796
/// @param metadata The KeyValueMetadata from file footer.
8897
/// @param compression Compression codec name.
@@ -139,14 +148,6 @@ class MapSharedShreddingUtils {
139148
static Result<std::map<std::string, int32_t>> BuildColumnToNumColumns(
140149
const std::vector<std::string>& shredding_field_names, const CoreOptions& options);
141150

142-
/// Serializes shredding metadata and appends entries to an existing KeyValueMetadata.
143-
/// @param field_meta The field-level shredding metadata to serialize.
144-
/// @param compression Compression codec name for field_dict compression.
145-
/// @param[out] metadata The KeyValueMetadata to append entries to.
146-
static Status SerializeMetadata(const MapSharedShreddingFieldMeta& field_meta,
147-
const std::string& compression,
148-
arrow::KeyValueMetadata* metadata);
149-
150151
/// Builds the physical Arrow type for one shredding MAP column.
151152
/// @param value_type The value type of the original MAP.
152153
/// @param num_columns Number of physical columns K.

‎src/paimon/common/data/shredding/map_shredding_defs.h‎

Lines changed: 2 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,9 @@
1717
#pragma once
1818

1919
#include <cstdint>
20-
#include <map>
21-
#include <set>
2220
#include <string>
23-
#include <vector>
21+
22+
#include "paimon/data/shredding/map_shared_shredding_schema_utils.h"
2423

2524
namespace paimon {
2625

@@ -71,27 +70,4 @@ struct MapSharedShreddingDefine {
7170
}
7271
};
7372

74-
/// Parsed file-level meta for one shared-shredding MAP column.
75-
struct MapSharedShreddingFieldMeta {
76-
/// field_name -> field_id
77-
std::map<std::string, int32_t> name_to_id;
78-
/// field_id -> set of physical column indices S
79-
std::map<int32_t, std::vector<int32_t>> field_to_columns;
80-
/// Set of field_ids that ever spilled into __overflow
81-
std::set<int32_t> overflow_field_set;
82-
/// Number of physical columns K in this file
83-
int32_t num_columns = 0;
84-
/// Maximum row width observed in this file
85-
int32_t max_row_width = 0;
86-
87-
bool operator==(const MapSharedShreddingFieldMeta& other) const {
88-
if (this == &other) {
89-
return true;
90-
}
91-
return name_to_id == other.name_to_id && field_to_columns == other.field_to_columns &&
92-
overflow_field_set == other.overflow_field_set && num_columns == other.num_columns &&
93-
max_row_width == other.max_row_width;
94-
}
95-
};
96-
9773
} // namespace paimon

0 commit comments

Comments
 (0)