Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions include/paimon/defs.h
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,15 @@ struct PAIMON_EXPORT Options {
/// "ignore-delete" - Whether to ignore delete records. Default value is "false".
static const char IGNORE_DELETE[];

/// "first-row.ignore-delete" deprecated as a fallback for `IGNORE_DELETE`.
static const char FALLBACK_FIRST_ROW_IGNORE_DELETE[];

/// "deduplicate.ignore-delete" deprecated as a fallback for `IGNORE_DELETE`.
static const char FALLBACK_DEDUPLICATE_IGNORE_DELETE[];

/// "partial-update.ignore-delete" deprecated as a fallback for `IGNORE_DELETE`.
static const char FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE[];
Comment thread
lxy-9602 marked this conversation as resolved.
Resolved

/// "fields.default-aggregate-function" - Default aggregate function of all fields for
/// partial-update and aggregate merge function.
static const char FIELDS_DEFAULT_AGG_FUNC[];
Expand Down
3 changes: 3 additions & 0 deletions src/paimon/common/defs.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,9 @@ const char Options::SEQUENCE_FIELD_SORT_ORDER[] = "sequence.field.sort-order";
const char Options::MERGE_ENGINE[] = "merge-engine";
const char Options::SORT_ENGINE[] = "sort-engine";
const char Options::IGNORE_DELETE[] = "ignore-delete";
const char Options::FALLBACK_FIRST_ROW_IGNORE_DELETE[] = "first-row.ignore-delete";
const char Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE[] = "deduplicate.ignore-delete";
const char Options::FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE[] = "partial-update.ignore-delete";
const char Options::FIELDS_DEFAULT_AGG_FUNC[] = "fields.default-aggregate-function";
const char Options::DELETION_VECTORS_ENABLED[] = "deletion-vectors.enabled";
const char Options::DELETION_VECTOR_INDEX_FILE_TARGET_SIZE[] =
Expand Down
17 changes: 15 additions & 2 deletions src/paimon/core/core_options.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -708,8 +708,21 @@ struct CoreOptions::Impl {
PAIMON_RETURN_NOT_OK(parser.ParseSortEngine(&sort_engine));
// Parse merge-engine - merge engine for primary key table, default "deduplicate"
PAIMON_RETURN_NOT_OK(parser.ParseMergeEngine(&merge_engine));
// Parse ignore-delete - whether to ignore delete records, default false
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::IGNORE_DELETE, &ignore_delete));
// Parse ignore-delete - whether to ignore delete records, default false.
// Java CoreOptions declares first-row.ignore-delete, deduplicate.ignore-delete
// and partial-update.ignore-delete as fallback keys, checked in that order only
// when ignore-delete itself is absent.
std::optional<bool> ignore_delete_value;
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::IGNORE_DELETE, &ignore_delete_value));
for (const char* fallback_key : {Options::FALLBACK_FIRST_ROW_IGNORE_DELETE,
Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE,
Options::FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE}) {
if (ignore_delete_value.has_value()) {
break;
}
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(fallback_key, &ignore_delete_value));
}
ignore_delete = ignore_delete_value.value_or(false);
// Parse fields.default-aggregate-function - default agg function for partial-update
PAIMON_RETURN_NOT_OK(parser.Parse(Options::FIELDS_DEFAULT_AGG_FUNC, &field_default_func));
// Parse changelog-producer - whether to double write to a changelog file, default "none"
Expand Down
44 changes: 44 additions & 0 deletions src/paimon/core/core_options_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -981,6 +981,50 @@ TEST(CoreOptionsTest, TestFallback) {
}
}

TEST(CoreOptionsTest, TestIgnoreDeleteFallbackKeys) {
{
// Tables written by Java may carry first-row.ignore-delete instead of ignore-delete.
ASSERT_OK_AND_ASSIGN(
CoreOptions options,
CoreOptions::FromMap({{Options::MERGE_ENGINE, "first-row"},
{Options::FALLBACK_FIRST_ROW_IGNORE_DELETE, "true"}}));
ASSERT_TRUE(options.IgnoreDelete());
}
{
ASSERT_OK_AND_ASSIGN(
CoreOptions options,
CoreOptions::FromMap({{Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE, "true"}}));
ASSERT_TRUE(options.IgnoreDelete());
}
{
ASSERT_OK_AND_ASSIGN(
CoreOptions options,
CoreOptions::FromMap({{Options::FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE, "true"}}));
ASSERT_TRUE(options.IgnoreDelete());
}
{
// The primary key takes precedence over fallback keys, matching Java CoreOptions.
ASSERT_OK_AND_ASSIGN(
CoreOptions options,
CoreOptions::FromMap({{Options::IGNORE_DELETE, "false"},
{Options::FALLBACK_FIRST_ROW_IGNORE_DELETE, "true"}}));
ASSERT_FALSE(options.IgnoreDelete());
}
{
// Fallback keys are checked in declaration order.
ASSERT_OK_AND_ASSIGN(
CoreOptions options,
CoreOptions::FromMap({{Options::FALLBACK_FIRST_ROW_IGNORE_DELETE, "false"},
{Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE, "true"}}));
ASSERT_FALSE(options.IgnoreDelete());
Comment thread
lxy-9602 marked this conversation as resolved.
Resolved
}
{
ASSERT_NOK_WITH_MSG(
CoreOptions::FromMap({{Options::FALLBACK_FIRST_ROW_IGNORE_DELETE, "invalid"}}),
"Invalid Config [first-row.ignore-delete: invalid]");
}
}

TEST(CoreOptionsTest, TestMapStorageLayout) {
// Test shared-shredding layout configured for a specific column
{
Expand Down
56 changes: 56 additions & 0 deletions src/paimon/core/utils/primary_key_table_utils_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,23 @@

#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <utility>

#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/types/row_kind.h"
#include "paimon/common/utils/fields_comparator.h"
#include "paimon/core/core_options.h"
#include "paimon/core/key_value.h"
#include "paimon/core/mergetree/compact/merge_function.h"
#include "paimon/defs.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/status.h"
#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/key_value_checker.h"
#include "paimon/testing/utils/testharness.h"

namespace paimon::test {
Expand Down Expand Up @@ -54,4 +62,52 @@ TEST(PrimaryKeyTableUtilsTest, TestCreateSequenceFieldsComparator) {
}
}

TEST(PrimaryKeyTableUtilsTest, TestCreateFirstRowMergeFunctionWithIgnoreDelete) {
auto pool = GetDefaultPool();
auto value_schema = arrow::schema({arrow::field("v0", arrow::int32())});

// ignore-delete can also be configured through the merge-engine-specific
// first-row.ignore-delete key, which must reach FirstRowMergeFunction the same way.
for (const char* ignore_delete_key :
{Options::IGNORE_DELETE, Options::FALLBACK_FIRST_ROW_IGNORE_DELETE}) {
ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap({{Options::MERGE_ENGINE, "first-row"},
{ignore_delete_key, "true"}}));
ASSERT_OK_AND_ASSIGN(
std::unique_ptr<MergeFunction> merge_function,
PrimaryKeyTableUtils::CreateMergeFunction(value_schema, {"k0"}, core_options));
merge_function->Reset();

KeyValue insert_kv(RowKind::Insert(), /*sequence_number=*/0, /*level=*/0, /*key=*/
BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
/*value=*/BinaryRowGenerator::GenerateRowPtr({100}, pool.get()));
KeyValue delete_kv(RowKind::Delete(), /*sequence_number=*/1, /*level=*/0, /*key=*/
BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
/*value=*/BinaryRowGenerator::GenerateRowPtr({200}, pool.get()));
ASSERT_OK(merge_function->Add(std::move(insert_kv)));
ASSERT_OK(merge_function->Add(std::move(delete_kv)));

ASSERT_OK_AND_ASSIGN(std::optional<KeyValue> result_kv, merge_function->GetResult());
ASSERT_TRUE(result_kv.has_value());
KeyValue expected(RowKind::Insert(), /*sequence_number=*/0, /*level=*/0, /*key=*/
BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
/*value=*/BinaryRowGenerator::GenerateRowPtr({100}, pool.get()));
KeyValueChecker::CheckResult(expected, result_kv.value(), /*key_arity=*/1,
/*value_arity=*/1);
}

// Without the option, the first-row merge engine still rejects retract records.
ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap({{Options::MERGE_ENGINE, "first-row"}}));
ASSERT_OK_AND_ASSIGN(
std::unique_ptr<MergeFunction> merge_function,
PrimaryKeyTableUtils::CreateMergeFunction(value_schema, {"k0"}, core_options));
merge_function->Reset();
KeyValue delete_kv(RowKind::Delete(), /*sequence_number=*/0, /*level=*/0, /*key=*/
BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
/*value=*/BinaryRowGenerator::GenerateRowPtr({100}, pool.get()));
ASSERT_NOK_WITH_MSG(merge_function->Add(std::move(delete_kv)),
"First row merge engine can not accept DELETE/UPDATE_BEFORE records");
}

} // namespace paimon::test
Loading