Skip to content

Commit e3123ae

Browse files
lxy-9602lxy-9602
authored
test: add map shared-shredding tests for compaction and alter table and predicates (alibaba#390)
1 parent 27bd4b2 commit e3123ae

6 files changed

Lines changed: 1678 additions & 375 deletions

‎src/paimon/core/schema/schema_validation_test.cpp‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -910,6 +910,26 @@ TEST(SchemaValidationTest, TestMapStorageLayout) {
910910
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
911911
"not MAP<STRING, T>");
912912
}
913+
// Invalid: nested MAP paths are not shared-shredding columns; only top-level columns are
914+
// addressable by fields.<column>.map.storage-layout.
915+
{
916+
auto payload = arrow::field(
917+
"payload",
918+
arrow::struct_({arrow::field("attrs", arrow::map(arrow::utf8(), arrow::int64()))}));
919+
arrow::FieldVector fields = {f0, f1, payload};
920+
auto schema = arrow::schema(fields);
921+
std::map<std::string, std::string> options = {
922+
{Options::BUCKET, "2"},
923+
{Options::BUCKET_KEY, "f0"},
924+
{"fields.payload.attrs.map.storage-layout", "shared-shredding"}};
925+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
926+
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
927+
/*primary_keys=*/{"f0", "f1"}, options));
928+
ASSERT_NOK_WITH_MSG(
929+
SchemaValidation::ValidateTableSchema(*table_schema),
930+
"Column 'payload.attrs' is configured with map.storage-layout but does not exist in "
931+
"table schema.");
932+
}
913933
// Valid: default layout on a MAP column
914934
{
915935
arrow::FieldVector fields = {f0, f1, f2};

‎test/inte/append_compaction_inte_test.cpp‎

Lines changed: 247 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,15 @@
1919
#include <string>
2020
#include <vector>
2121

22+
#include "arrow/api.h"
2223
#include "arrow/c/bridge.h"
2324
#include "gtest/gtest.h"
2425
#include "paimon/commit_context.h"
2526
#include "paimon/common/data/binary_row.h"
27+
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
2628
#include "paimon/common/factories/io_hook.h"
29+
#include "paimon/common/types/data_field.h"
30+
#include "paimon/common/utils/path_util.h"
2731
#include "paimon/common/utils/scope_guard.h"
2832
#include "paimon/core/append/bucketed_append_compact_manager.h"
2933
#include "paimon/core/io/data_file_meta.h"
@@ -34,10 +38,14 @@
3438
#include "paimon/executor.h"
3539
#include "paimon/file_store_commit.h"
3640
#include "paimon/file_store_write.h"
41+
#include "paimon/format/file_format_factory.h"
42+
#include "paimon/read_context.h"
3743
#include "paimon/result.h"
44+
#include "paimon/table/source/table_read.h"
3845
#include "paimon/testing/utils/binary_row_generator.h"
3946
#include "paimon/testing/utils/data_generator.h"
4047
#include "paimon/testing/utils/io_exception_helper.h"
48+
#include "paimon/testing/utils/read_result_collector.h"
4149
#include "paimon/testing/utils/test_helper.h"
4250
#include "paimon/testing/utils/testharness.h"
4351
#include "paimon/write_context.h"
@@ -245,6 +253,245 @@ TEST_P(AppendCompactionInteTest, TestAppendTableStreamWriteFullCompaction) {
245253
}
246254
}
247255

256+
TEST_P(AppendCompactionInteTest, TestAppendTableStreamWriteFullCompactionWithMapSharedShredding) {
257+
auto file_format = GetParam();
258+
if (file_format != "parquet" && file_format != "orc") {
259+
return;
260+
}
261+
262+
auto dir = UniqueTestDirectory::Create();
263+
ASSERT_TRUE(dir);
264+
auto map_type = arrow::map(arrow::utf8(), arrow::int64());
265+
arrow::FieldVector fields = {
266+
arrow::field("id", arrow::int32()),
267+
arrow::field("tags", map_type),
268+
};
269+
auto schema = arrow::schema(fields);
270+
271+
std::map<std::string, std::string> options = {
272+
{Options::FILE_FORMAT, file_format},
273+
{Options::BUCKET, "1"},
274+
{Options::BUCKET_KEY, "id"},
275+
{Options::FILE_SYSTEM, "local"},
276+
{"fields.tags.map.storage-layout", "shared-shredding"},
277+
{"fields.tags.map.shared-shredding.max-columns", "64"},
278+
};
279+
ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
280+
/*primary_keys=*/{}, options,
281+
/*is_streaming_mode=*/true));
282+
283+
ASSERT_OK_AND_ASSIGN(auto batch_0,
284+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
285+
R"([
286+
[1, [["a", 10], ["b", 20]]],
287+
[2, [["c", 30]]]
288+
])",
289+
/*partition_map=*/{}, /*bucket=*/0, {}));
290+
int64_t commit_identifier = 0;
291+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_0), commit_identifier++,
292+
/*expected_commit_messages=*/std::nullopt));
293+
294+
ASSERT_OK_AND_ASSIGN(auto batch_1,
295+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
296+
R"([
297+
[3, [["a", 40], ["d", 50]]],
298+
[4, null]
299+
])",
300+
/*partition_map=*/{}, /*bucket=*/0, {}));
301+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_1), commit_identifier++,
302+
/*expected_commit_messages=*/std::nullopt));
303+
304+
ASSERT_OK_AND_ASSIGN(auto batch_2,
305+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
306+
R"([
307+
[5, [["e", 60], ["f", 70], ["g", 80], ["h", 90]]]
308+
])",
309+
/*partition_map=*/{}, /*bucket=*/0, {}));
310+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), commit_identifier++,
311+
/*expected_commit_messages=*/std::nullopt));
312+
313+
ASSERT_OK(helper->write_->Compact(/*partition=*/{}, /*bucket=*/0,
314+
/*full_compaction=*/true));
315+
ASSERT_OK_AND_ASSIGN(
316+
std::vector<std::shared_ptr<CommitMessage>> commit_messages,
317+
helper->write_->PrepareCommit(/*wait_compaction=*/true, commit_identifier));
318+
ASSERT_FALSE(commit_messages.empty());
319+
ASSERT_OK(helper->commit_->Commit(commit_messages, commit_identifier));
320+
ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot, helper->LatestSnapshot());
321+
ASSERT_TRUE(snapshot);
322+
ASSERT_EQ(Snapshot::CommitKind::Compact(), snapshot.value().GetCommitKind());
323+
324+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
325+
helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
326+
ASSERT_EQ(data_splits.size(), 1);
327+
{
328+
// check adaptive k
329+
auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(data_splits[0]);
330+
ASSERT_TRUE(data_split);
331+
ASSERT_EQ(data_split->DataFiles().size(), 1);
332+
auto compact_file = data_split->DataFiles()[0];
333+
std::string compact_file_path =
334+
PathUtil::JoinPath(data_split->BucketPath(), compact_file->file_name);
335+
ASSERT_OK_AND_ASSIGN(auto unique_input_stream,
336+
dir->GetFileSystem()->Open(compact_file_path));
337+
std::shared_ptr<InputStream> input_stream(std::move(unique_input_stream));
338+
ASSERT_OK_AND_ASSIGN(auto file_format_obj, FileFormatFactory::Get(file_format, options));
339+
ASSERT_OK_AND_ASSIGN(auto reader_builder, file_format_obj->CreateReaderBuilder(10));
340+
ASSERT_OK_AND_ASSIGN(auto reader, reader_builder->Build(input_stream));
341+
ASSERT_OK_AND_ASSIGN(auto c_file_schema, reader->GetFileSchema());
342+
auto file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOrDie();
343+
auto tags_field = file_schema->GetFieldByName("tags");
344+
ASSERT_TRUE(tags_field);
345+
ASSERT_TRUE(tags_field->metadata());
346+
ASSERT_OK_AND_ASSIGN(
347+
auto tags_meta,
348+
MapSharedShreddingUtils::DeserializeMetadata(
349+
tags_field->metadata()->Copy(), MapSharedShreddingDefine::kDefaultDictCompression));
350+
ASSERT_EQ(4, tags_meta.num_columns);
351+
ASSERT_EQ(4, tags_meta.max_row_width);
352+
}
353+
{
354+
// recall all fields
355+
arrow::FieldVector fields_with_row_kind = fields;
356+
fields_with_row_kind.insert(fields_with_row_kind.begin(),
357+
arrow::field("_VALUE_KIND", arrow::int8()));
358+
auto data_type = arrow::struct_(fields_with_row_kind);
359+
ASSERT_OK_AND_ASSIGN(bool success, helper->ReadAndCheckResult(data_type, data_splits,
360+
R"([
361+
[0, 1, [["a", 10], ["b", 20]]],
362+
[0, 2, [["c", 30]]],
363+
[0, 3, [["a", 40], ["d", 50]]],
364+
[0, 4, null],
365+
[0, 5, [["e", 60], ["f", 70], ["g", 80], ["h", 90]]]
366+
])"));
367+
ASSERT_TRUE(success);
368+
}
369+
{
370+
// recall only "a,f" sub-key in map
371+
auto selected_keys_meta =
372+
arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,f"});
373+
auto read_schema = arrow::schema({
374+
arrow::field("id", arrow::int32()),
375+
arrow::field("tags", map_type)->WithMetadata(selected_keys_meta),
376+
});
377+
auto c_schema = std::make_unique<ArrowSchema>();
378+
ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());
379+
380+
ReadContextBuilder read_context_builder(PathUtil::JoinPath(dir->Str(), "foo.db/bar"));
381+
read_context_builder.SetOptions(options).SetReadSchema(std::move(c_schema));
382+
ASSERT_OK_AND_ASSIGN(auto read_context, read_context_builder.Finish());
383+
ASSERT_OK_AND_ASSIGN(auto table_read, TableRead::Create(std::move(read_context)));
384+
ASSERT_OK_AND_ASSIGN(auto batch_reader, table_read->CreateReader(data_splits));
385+
ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(batch_reader.get()));
386+
387+
auto expected_type = arrow::struct_({
388+
arrow::field("_VALUE_KIND", arrow::int8()),
389+
arrow::field("id", arrow::int32()),
390+
arrow::field("tags", map_type),
391+
});
392+
auto expected = arrow::ipc::internal::json::ArrayFromJSON(expected_type, R"([
393+
[0, 1, [["a", 10]]],
394+
[0, 2, []],
395+
[0, 3, [["a", 40]]],
396+
[0, 4, null],
397+
[0, 5, [["f", 70]]]
398+
])")
399+
.ValueOrDie();
400+
auto expected_chunked = std::make_shared<arrow::ChunkedArray>(expected);
401+
ASSERT_TRUE(expected_chunked->Equals(actual))
402+
<< "actual=" << actual->ToString() << "\nexpected=" << expected_chunked->ToString();
403+
}
404+
}
405+
406+
TEST_P(AppendCompactionInteTest,
407+
TestOrcAppendTableFullCompactionWithMapSharedShreddingStringValue) {
408+
auto file_format = GetParam();
409+
if (file_format != "orc") {
410+
return;
411+
}
412+
413+
auto dir = UniqueTestDirectory::Create();
414+
ASSERT_TRUE(dir);
415+
auto map_type = arrow::map(arrow::utf8(), arrow::utf8());
416+
arrow::FieldVector fields = {
417+
arrow::field("id", arrow::int32()),
418+
arrow::field("tags", map_type),
419+
};
420+
auto schema = arrow::schema(fields);
421+
422+
std::map<std::string, std::string> options = {
423+
{Options::FILE_FORMAT, "orc"},
424+
{Options::BUCKET, "1"},
425+
{Options::BUCKET_KEY, "id"},
426+
{Options::FILE_SYSTEM, "local"},
427+
{"orc.read.enable-lazy-decoding", "true"},
428+
{"orc.dictionary-key-size-threshold", "1"},
429+
{"fields.tags.map.storage-layout", "shared-shredding"},
430+
{"fields.tags.map.shared-shredding.max-columns", "1"},
431+
};
432+
ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
433+
/*primary_keys=*/{}, options,
434+
/*is_streaming_mode=*/true));
435+
436+
int64_t commit_identifier = 0;
437+
ASSERT_OK_AND_ASSIGN(auto batch_0,
438+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
439+
R"([
440+
[1, [["a", "shared"], ["b", "hot"]]],
441+
[2, [["c", "shared"]]]
442+
])",
443+
/*partition_map=*/{}, /*bucket=*/0, {}));
444+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_0), commit_identifier++,
445+
/*expected_commit_messages=*/std::nullopt));
446+
447+
ASSERT_OK_AND_ASSIGN(auto batch_1,
448+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
449+
R"([
450+
[3, [["a", "shared"], ["d", "hot"]]],
451+
[4, null]
452+
])",
453+
/*partition_map=*/{}, /*bucket=*/0, {}));
454+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_1), commit_identifier++,
455+
/*expected_commit_messages=*/std::nullopt));
456+
457+
ASSERT_OK_AND_ASSIGN(auto batch_2,
458+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
459+
R"([
460+
[5, [["e", "shared"], ["f", "hot"], ["g", "shared"]]]
461+
])",
462+
/*partition_map=*/{}, /*bucket=*/0, {}));
463+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), commit_identifier++,
464+
/*expected_commit_messages=*/std::nullopt));
465+
466+
ASSERT_OK(helper->write_->Compact(/*partition=*/{}, /*bucket=*/0,
467+
/*full_compaction=*/true));
468+
ASSERT_OK_AND_ASSIGN(
469+
std::vector<std::shared_ptr<CommitMessage>> commit_messages,
470+
helper->write_->PrepareCommit(/*wait_compaction=*/true, commit_identifier));
471+
ASSERT_FALSE(commit_messages.empty());
472+
ASSERT_OK(helper->commit_->Commit(commit_messages, commit_identifier));
473+
ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot, helper->LatestSnapshot());
474+
ASSERT_TRUE(snapshot);
475+
ASSERT_EQ(Snapshot::CommitKind::Compact(), snapshot.value().GetCommitKind());
476+
477+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
478+
helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
479+
ASSERT_EQ(data_splits.size(), 1);
480+
arrow::FieldVector fields_with_row_kind = fields;
481+
fields_with_row_kind.insert(fields_with_row_kind.begin(),
482+
arrow::field("_VALUE_KIND", arrow::int8()));
483+
auto data_type = arrow::struct_(fields_with_row_kind);
484+
ASSERT_OK_AND_ASSIGN(bool success, helper->ReadAndCheckResult(data_type, data_splits,
485+
R"([
486+
[0, 1, [["a", "shared"], ["b", "hot"]]],
487+
[0, 2, [["c", "shared"]]],
488+
[0, 3, [["a", "shared"], ["d", "hot"]]],
489+
[0, 4, null],
490+
[0, 5, [["e", "shared"], ["f", "hot"], ["g", "shared"]]]
491+
])"));
492+
ASSERT_TRUE(success);
493+
}
494+
248495
TEST_P(AppendCompactionInteTest, TestAppendTableStreamWriteFullCompactionWithDv) {
249496
auto dir = UniqueTestDirectory::Create();
250497
ASSERT_TRUE(dir);

0 commit comments

Comments
 (0)