Skip to content

Commit baaf4c1

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

6 files changed

Lines changed: 412 additions & 34 deletions

File tree

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
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+
#pragma once
18+
19+
#include <cstdint>
20+
#include <map>
21+
#include <memory>
22+
#include <set>
23+
#include <string>
24+
#include <vector>
25+
26+
#include "paimon/arrow/abi.h"
27+
#include "paimon/result.h"
28+
#include "paimon/status.h"
29+
#include "paimon/visibility.h"
30+
31+
namespace paimon {
32+
33+
/// Parsed file-level meta for one shared-shredding MAP column.
34+
struct PAIMON_EXPORT MapSharedShreddingFieldMeta {
35+
/// field_name -> field_id
36+
std::map<std::string, int32_t> name_to_id;
37+
/// field_id -> set of physical column indices S
38+
std::map<int32_t, std::vector<int32_t>> field_to_columns;
39+
/// Set of field_ids that ever spilled into __overflow
40+
std::set<int32_t> overflow_field_set;
41+
/// Number of physical columns K in this file
42+
int32_t num_columns = 0;
43+
/// Maximum row width observed in this file
44+
int32_t max_row_width = 0;
45+
46+
bool operator==(const MapSharedShreddingFieldMeta& other) const {
47+
if (this == &other) {
48+
return true;
49+
}
50+
return name_to_id == other.name_to_id && field_to_columns == other.field_to_columns &&
51+
overflow_field_set == other.overflow_field_set && num_columns == other.num_columns &&
52+
max_row_width == other.max_row_width;
53+
}
54+
};
55+
56+
class PAIMON_EXPORT MapSharedShreddingSchemaUtils {
57+
public:
58+
MapSharedShreddingSchemaUtils() = delete;
59+
~MapSharedShreddingSchemaUtils() = delete;
60+
61+
/// Converts a logical schema to a physical schema by replacing shredding MAP columns
62+
/// with their physical Struct representation.
63+
/// @param logical_schema The original Arrow C schema with MAP<STRING, T> columns.
64+
/// Ownership of schema resources is transferred to this method.
65+
/// @param field_to_num_columns Map from field name to its physical column count K.
66+
/// Each shredding column can have its own width.
67+
/// @return The exported Arrow C schema for file writing.
68+
static Result<std::unique_ptr<::ArrowSchema>> LogicalToPhysicalSchema(
69+
std::unique_ptr<::ArrowSchema> logical_schema,
70+
const std::map<std::string, int32_t>& field_to_num_columns);
71+
72+
/// Attaches shared-shredding metadata to fields in a physical schema.
73+
/// @param physical_schema The Arrow C physical schema whose fields should receive metadata.
74+
/// Ownership of schema resources is transferred to this method.
75+
/// @param field_name_to_meta Map from physical field name to its shared-shredding metadata.
76+
/// @param compression Compression codec name for field_dict serialization.
77+
/// @return A new Arrow C schema with shared-shredding metadata attached to matching fields.
78+
static Result<std::unique_ptr<::ArrowSchema>> AttachMetadataToSchema(
79+
std::unique_ptr<::ArrowSchema> physical_schema,
80+
const std::map<std::string, MapSharedShreddingFieldMeta>& field_name_to_meta,
81+
const std::string& compression);
82+
83+
/// Extracts shared-shredding metadata from a named field in a physical schema.
84+
/// @param physical_schema The Arrow C physical schema that contains the target field.
85+
/// Ownership of schema resources is transferred to this method.
86+
/// @param field_name The physical field name whose metadata should be extracted.
87+
/// @param compression Compression codec name for field_dict deserialization.
88+
/// @return Parsed shared-shredding metadata for the field.
89+
static Result<MapSharedShreddingFieldMeta> ExtractMetadataFromField(
90+
std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name,
91+
const std::string& compression);
92+
};
93+
94+
} // 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
Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
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+
// Type guard: verify all fields in field_to_num_columns are actually MAP type
37+
for (const auto& [field_name, num_columns] : field_to_num_columns) {
38+
std::shared_ptr<arrow::Field> field = arrow_logical_schema->GetFieldByName(field_name);
39+
if (!field) {
40+
return Status::Invalid(
41+
fmt::format("Field '{}' not found in logical schema.", field_name));
42+
}
43+
if (field->type()->id() != arrow::Type::MAP) {
44+
return Status::Invalid(
45+
fmt::format("Field '{}' is expected to be MAP type, but got '{}'.", field_name,
46+
field->type()->name()));
47+
}
48+
}
49+
std::shared_ptr<arrow::Schema> physical_schema =
50+
MapSharedShreddingUtils::LogicalToPhysicalSchema(arrow_logical_schema,
51+
field_to_num_columns);
52+
auto c_schema = std::make_unique<::ArrowSchema>();
53+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*physical_schema, c_schema.get()));
54+
return c_schema;
55+
}
56+
57+
Result<std::unique_ptr<::ArrowSchema>> MapSharedShreddingSchemaUtils::AttachMetadataToSchema(
58+
std::unique_ptr<::ArrowSchema> physical_schema,
59+
const std::map<std::string, MapSharedShreddingFieldMeta>& field_name_to_meta,
60+
const std::string& compression) {
61+
if (!physical_schema) {
62+
return Status::Invalid("physical schema is null");
63+
}
64+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_physical_schema,
65+
arrow::ImportSchema(physical_schema.get()));
66+
67+
arrow::FieldVector updated_fields = arrow_physical_schema->fields();
68+
for (const auto& [field_name, field_meta] : field_name_to_meta) {
69+
int32_t field_index = arrow_physical_schema->GetFieldIndex(field_name);
70+
if (field_index < 0) {
71+
return Status::Invalid(fmt::format(
72+
"Shared-shredding field '{}' not found in physical schema.", field_name));
73+
}
74+
75+
const auto& field = arrow_physical_schema->field(field_index);
76+
auto metadata = field->metadata() ? field->metadata()->Copy()
77+
: std::make_shared<arrow::KeyValueMetadata>();
78+
PAIMON_RETURN_NOT_OK(
79+
MapSharedShreddingUtils::SerializeMetadata(field_meta, compression, metadata.get()));
80+
updated_fields[field_index] = field->WithMetadata(metadata);
81+
}
82+
83+
auto updated_schema =
84+
arrow::schema(std::move(updated_fields), arrow_physical_schema->metadata());
85+
auto c_schema = std::make_unique<::ArrowSchema>();
86+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*updated_schema, c_schema.get()));
87+
return c_schema;
88+
}
89+
90+
Result<MapSharedShreddingFieldMeta> MapSharedShreddingSchemaUtils::ExtractMetadataFromField(
91+
std::unique_ptr<::ArrowSchema> physical_schema, const std::string& field_name,
92+
const std::string& compression) {
93+
if (!physical_schema) {
94+
return Status::Invalid("physical schema is null");
95+
}
96+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_physical_schema,
97+
arrow::ImportSchema(physical_schema.get()));
98+
const auto& field = arrow_physical_schema->GetFieldByName(field_name);
99+
if (!field) {
100+
return Status::Invalid(
101+
fmt::format("Shared-shredding field '{}' not found in physical schema.", field_name));
102+
}
103+
104+
auto metadata =
105+
field->metadata() ? field->metadata()->Copy() : std::shared_ptr<arrow::KeyValueMetadata>();
106+
return MapSharedShreddingUtils::DeserializeMetadata(metadata, compression);
107+
}
108+
109+
} // namespace paimon

0 commit comments

Comments
 (0)