Skip to content

Commit a710355

Browse files
dalingmengdalingmeng
committed
Expose shared-shredding schema metadata utilities
1 parent ffbda4f commit a710355

6 files changed

Lines changed: 269 additions & 37 deletions

File tree

‎include/paimon/shredding/map_shared_shredding_schema_utils.h‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
#pragma once
2+
3+
#include <cstdint>
4+
#include <map>
5+
#include <memory>
6+
#include <set>
7+
#include <string>
8+
#include <vector>
9+
10+
#include "paimon/result.h"
11+
#include "paimon/status.h"
12+
#include "paimon/visibility.h"
13+
14+
namespace arrow {
15+
class Field;
16+
class Schema;
17+
} // namespace arrow
18+
19+
namespace paimon {
20+
21+
/// Parsed file-level meta for one shared-shredding MAP column.
22+
struct PAIMON_EXPORT MapSharedShreddingFieldMeta {
23+
/// field_name -> field_id
24+
std::map<std::string, int32_t> name_to_id;
25+
/// field_id -> set of physical column indices S
26+
std::map<int32_t, std::vector<int32_t>> field_to_columns;
27+
/// Set of field_ids that ever spilled into __overflow
28+
std::set<int32_t> overflow_field_set;
29+
/// Number of physical columns K in this file
30+
int32_t num_columns = 0;
31+
/// Maximum row width observed in this file
32+
int32_t max_row_width = 0;
33+
34+
bool operator==(const MapSharedShreddingFieldMeta& other) const {
35+
if (this == &other) {
36+
return true;
37+
}
38+
return name_to_id == other.name_to_id && field_to_columns == other.field_to_columns &&
39+
overflow_field_set == other.overflow_field_set && num_columns == other.num_columns &&
40+
max_row_width == other.max_row_width;
41+
}
42+
};
43+
44+
class PAIMON_EXPORT MapSharedShreddingSchemaUtils {
45+
public:
46+
MapSharedShreddingSchemaUtils() = delete;
47+
~MapSharedShreddingSchemaUtils() = delete;
48+
49+
/// Per-row field mapping sub-column name.
50+
static constexpr const char* kFieldMapping = "__field_mapping";
51+
/// Overflow sub-column name.
52+
static constexpr const char* kOverflow = "__overflow";
53+
54+
/// Returns the name of the i-th physical value sub-column.
55+
static std::string PhysicalColumnName(int32_t index) {
56+
return "__col_" + std::to_string(index);
57+
}
58+
59+
static Result<std::shared_ptr<arrow::Schema>> AttachMetadataToSchema(
60+
const std::shared_ptr<arrow::Schema>& physical_schema,
61+
const std::map<std::string, MapSharedShreddingFieldMeta>& field_name_to_meta,
62+
const std::string& compression);
63+
64+
static Result<MapSharedShreddingFieldMeta> ExtractMetadataFromField(
65+
const std::shared_ptr<arrow::Field>& field, const std::string& compression);
66+
};
67+
68+
} // namespace paimon

‎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: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
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/shredding/map_shared_shredding_schema_utils.h"
18+
19+
#include "arrow/type.h"
20+
#include "arrow/util/key_value_metadata.h"
21+
#include "fmt/format.h"
22+
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
23+
24+
namespace paimon {
25+
26+
Result<std::shared_ptr<arrow::Schema>> MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
27+
const std::shared_ptr<arrow::Schema>& physical_schema,
28+
const std::map<std::string, MapSharedShreddingFieldMeta>& field_name_to_meta,
29+
const std::string& compression) {
30+
if (!physical_schema) {
31+
return Status::Invalid("physical schema is null");
32+
}
33+
34+
arrow::FieldVector updated_fields = physical_schema->fields();
35+
for (const auto& [field_name, field_meta] : field_name_to_meta) {
36+
int32_t field_index = physical_schema->GetFieldIndex(field_name);
37+
if (field_index < 0) {
38+
return Status::Invalid(fmt::format(
39+
"Shared-shredding field '{}' not found in physical schema.", field_name));
40+
}
41+
42+
const auto& field = physical_schema->field(field_index);
43+
auto metadata = field->metadata() ? field->metadata()->Copy()
44+
: std::make_shared<arrow::KeyValueMetadata>();
45+
PAIMON_RETURN_NOT_OK(
46+
MapSharedShreddingUtils::SerializeMetadata(field_meta, compression, metadata.get()));
47+
updated_fields[field_index] = field->WithMetadata(metadata);
48+
}
49+
50+
return arrow::schema(std::move(updated_fields), physical_schema->metadata());
51+
}
52+
53+
Result<MapSharedShreddingFieldMeta> MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
54+
const std::shared_ptr<arrow::Field>& field, const std::string& compression) {
55+
if (!field) {
56+
return Status::Invalid("field is null");
57+
}
58+
59+
auto metadata =
60+
field->metadata() ? field->metadata()->Copy() : std::shared_ptr<arrow::KeyValueMetadata>();
61+
return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression);
62+
}
63+
64+
} // namespace paimon

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

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
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/shredding/map_shared_shredding_schema_utils.h"
18+
19+
#include <string>
20+
21+
#include "arrow/type.h"
22+
#include "arrow/util/key_value_metadata.h"
23+
#include "gtest/gtest.h"
24+
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
25+
#include "paimon/testing/utils/testharness.h"
26+
27+
namespace paimon::test {
28+
29+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaBasic) {
30+
MapSharedShreddingFieldMeta tags_meta;
31+
tags_meta.name_to_id = {{"host", 0}, {"region", 1}};
32+
tags_meta.field_to_columns = {{0, {0}}, {1, {1}}};
33+
tags_meta.num_columns = 2;
34+
tags_meta.max_row_width = 2;
35+
36+
auto id_metadata = std::make_shared<arrow::KeyValueMetadata>();
37+
id_metadata->Append("paimon.field.id", "1");
38+
auto schema_metadata = std::make_shared<arrow::KeyValueMetadata>();
39+
schema_metadata->Append("schema.key", "schema.value");
40+
auto schema = arrow::schema(
41+
{arrow::field("id", arrow::int32(), true, id_metadata),
42+
arrow::field("tags", arrow::struct_({arrow::field("__field_mapping",
43+
arrow::list(arrow::int32()), true),
44+
arrow::field("__col_0", arrow::utf8(), true),
45+
arrow::field("__col_1", arrow::utf8(), true)}))},
46+
schema_metadata);
47+
48+
ASSERT_OK_AND_ASSIGN(auto updated_schema, MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
49+
schema, {{"tags", tags_meta}}, "none"));
50+
51+
ASSERT_TRUE(updated_schema->metadata()->Equals(*schema_metadata));
52+
ASSERT_TRUE(updated_schema->field(0)->metadata()->Equals(*id_metadata));
53+
54+
auto tags_metadata = updated_schema->GetFieldByName("tags")->metadata()->Copy();
55+
ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(tags_metadata));
56+
ASSERT_OK_AND_ASSIGN(auto deserialized,
57+
MapSharedShreddingUtils::DeserializeMetadata(tags_metadata, "none"));
58+
ASSERT_EQ(deserialized, tags_meta);
59+
}
60+
61+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaMissingField) {
62+
MapSharedShreddingFieldMeta tags_meta;
63+
tags_meta.name_to_id = {{"host", 0}};
64+
tags_meta.field_to_columns = {{0, {0}}};
65+
tags_meta.num_columns = 1;
66+
tags_meta.max_row_width = 1;
67+
68+
auto schema = arrow::schema({arrow::field("id", arrow::int32())});
69+
70+
ASSERT_NOK_WITH_MSG(MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
71+
schema, {{"tags", tags_meta}}, "none"),
72+
"Shared-shredding field 'tags' not found in physical schema.");
73+
}
74+
75+
TEST(MapSharedShreddingSchemaUtilsTest, AttachMetadataToSchemaPreservesExistingFieldMetadata) {
76+
MapSharedShreddingFieldMeta tags_meta;
77+
tags_meta.name_to_id = {{"host", 0}};
78+
tags_meta.field_to_columns = {{0, {0}}};
79+
tags_meta.num_columns = 1;
80+
tags_meta.max_row_width = 1;
81+
82+
auto tags_metadata = std::make_shared<arrow::KeyValueMetadata>();
83+
tags_metadata->Append("paimon.field.id", "7");
84+
tags_metadata->Append("description", "original tags field");
85+
auto schema = arrow::schema({arrow::field(
86+
"tags",
87+
arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
88+
arrow::field("__col_0", arrow::utf8(), true)}),
89+
true, tags_metadata)});
90+
91+
ASSERT_OK_AND_ASSIGN(auto updated_schema, MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
92+
schema, {{"tags", tags_meta}}, "none"));
93+
94+
auto updated_metadata = updated_schema->field(0)->metadata()->Copy();
95+
ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("paimon.field.id")), "7");
96+
ASSERT_EQ(updated_metadata->value(updated_metadata->FindKey("description")),
97+
"original tags field");
98+
ASSERT_TRUE(MapSharedShreddingUtils::HasShreddingMetadata(updated_metadata));
99+
}
100+
101+
TEST(MapSharedShreddingSchemaUtilsTest, ExtractMetadataFromFieldBasic) {
102+
MapSharedShreddingFieldMeta tags_meta;
103+
tags_meta.name_to_id = {{"host", 0}, {"region", 1}};
104+
tags_meta.field_to_columns = {{0, {0}}, {1, {1}}};
105+
tags_meta.num_columns = 2;
106+
tags_meta.max_row_width = 2;
107+
108+
auto metadata = std::make_shared<arrow::KeyValueMetadata>();
109+
ASSERT_OK(MapSharedShreddingUtils::SerializeMetadata(tags_meta, "none", metadata.get()));
110+
auto field = arrow::field(
111+
"tags",
112+
arrow::struct_({arrow::field("__field_mapping", arrow::list(arrow::int32()), true),
113+
arrow::field("__col_0", arrow::utf8(), true),
114+
arrow::field("__col_1", arrow::utf8(), true)}),
115+
true, metadata);
116+
117+
ASSERT_OK_AND_ASSIGN(auto parsed_meta,
118+
MapSharedShreddingSchemaUtils::ExtractMetadataFromField(field, "none"));
119+
ASSERT_EQ(parsed_meta, tags_meta);
120+
}
121+
122+
} // namespace paimon::test

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,14 @@ class MapSharedShreddingUtils {
8383

8484
// ---- Metadata serialization ----
8585

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

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-
150150
/// Builds the physical Arrow type for one shredding MAP column.
151151
/// @param value_type The value type of the original MAP.
152152
/// @param num_columns Number of physical columns K.

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

Lines changed: 5 additions & 29 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/shredding/map_shared_shredding_schema_utils.h"
2423

2524
namespace paimon {
2625

@@ -58,39 +57,16 @@ struct MapSharedShreddingDefine {
5857
// ---- Physical sub-column names ----
5958

6059
/// Per-row field mapping column name.
61-
static constexpr const char* kFieldMapping = "__field_mapping";
60+
static constexpr const char* kFieldMapping = MapSharedShreddingSchemaUtils::kFieldMapping;
6261
/// Overflow column name.
63-
static constexpr const char* kOverflow = "__overflow";
62+
static constexpr const char* kOverflow = MapSharedShreddingSchemaUtils::kOverflow;
6463

6564
/// Default compression codec for field_dict serialization.
6665
static constexpr const char* kDefaultDictCompression = "zstd";
6766

6867
/// Returns the name of the i-th physical column: "__col_0", "__col_1", etc.
6968
static std::string PhysicalColumnName(int32_t index) {
70-
return "__col_" + std::to_string(index);
71-
}
72-
};
73-
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;
69+
return MapSharedShreddingSchemaUtils::PhysicalColumnName(index);
9470
}
9571
};
9672

0 commit comments

Comments
 (0)