feat(core): support input and lookup changelog producers

include/paimon/defs.h

@@ -386,14 +386,34 @@ struct PAIMON_EXPORT Options {
     /// @note: bitmap64 dv is not supported.
     static const char DELETION_VECTOR_BITMAP64[];
 
-    ///  @note `CHANGELOG_PRODUCER` currently only support `none`
-    ///
     /// "changelog-producer" - Whether to double write to a changelog file. This changelog file
     /// keeps the details of data changes, it can be read directly during stream reads. This can be
     /// applied to tables with primary keys. Values can be "none", "input", "lookup",
     /// "full-compaction". Default value is "none".
+    /// @note C++ Paimon currently supports "none", "input", and "lookup".
     static const char CHANGELOG_PRODUCER[];
 
+    /// "changelog-producer.row-deduplicate" - Whether to generate update-before and update-after
+    /// changelog records when the row has not changed. This option is only valid for "lookup" or
+    /// "full-compaction" changelog producers. Default value is "false".
+    static const char CHANGELOG_PRODUCER_ROW_DEDUPLICATE[];
+
+    /// "changelog-producer.row-deduplicate-ignore-fields" - Comma-separated fields to ignore when
+    /// comparing rows for changelog deduplication. This option is only valid when
+    /// "changelog-producer.row-deduplicate" is "true".
+    static const char CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS[];
+
+    /// "changelog-file.prefix" - Specify the file name prefix of changelog files. Default value is
+    /// "changelog-".
+    static const char CHANGELOG_FILE_PREFIX[];
+
+    /// "changelog-file.format" - Specify the file format of changelog files. No default value.
+    static const char CHANGELOG_FILE_FORMAT[];
+
+    /// "changelog-file.compression" - Specify the compression of changelog files. No default
+    /// value.
+    static const char CHANGELOG_FILE_COMPRESSION[];
+
     /// "force-lookup" - Whether to force the use of lookup for compaction. Default value is
     /// "false".
     static const char FORCE_LOOKUP[];

src/paimon/CMakeLists.txt

@@ -288,6 +288,7 @@ set(PAIMON_CORE_SRCS
     core/io/vector_file_batch_reader.cpp
     core/io/file_index_evaluator.cpp
     core/io/key_value_data_file_record_reader.cpp
+    core/io/key_value_data_file_writer_factories.cpp
     core/io/key_value_data_file_writer_factory.cpp
     core/io/key_value_data_file_writer.cpp
     core/io/key_value_in_memory_record_reader.cpp
@@ -814,6 +815,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/compact/deduplicate_merge_function_test.cpp
                     core/mergetree/compact/first_row_merge_function_test.cpp
                     core/mergetree/compact/first_row_merge_function_wrapper_test.cpp
+                    core/mergetree/compact/internal_row_equalizer_test.cpp
                     core/mergetree/compact/interval_partition_test.cpp
                     core/mergetree/compact/lookup_changelog_merge_function_wrapper_test.cpp
                     core/mergetree/compact/lookup_merge_tree_compact_rewriter_test.cpp

src/paimon/common/defs.cpp

@@ -96,6 +96,12 @@ const char Options::DELETION_VECTOR_INDEX_FILE_TARGET_SIZE[] =
     "deletion-vector.index-file.target-size";
 const char Options::DELETION_VECTOR_BITMAP64[] = "deletion-vectors.bitmap64";
 const char Options::CHANGELOG_PRODUCER[] = "changelog-producer";
+const char Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE[] = "changelog-producer.row-deduplicate";
+const char Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS[] =
+    "changelog-producer.row-deduplicate-ignore-fields";
+const char Options::CHANGELOG_FILE_PREFIX[] = "changelog-file.prefix";
+const char Options::CHANGELOG_FILE_FORMAT[] = "changelog-file.format";
+const char Options::CHANGELOG_FILE_COMPRESSION[] = "changelog-file.compression";
 const char Options::FORCE_LOOKUP[] = "force-lookup";
 const char Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE[] =
     "partial-update.remove-record-on-delete";

src/paimon/common/utils/fields_comparator.cpp

@@ -85,68 +85,65 @@ Result<FieldsComparator::FieldComparatorFunc> FieldsComparator::CompareField(
     arrow::Type::type type = input_type->id();
     switch (type) {
         case arrow::Type::type::BOOL:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     bool lvalue = lhs.GetBoolean(field_idx);
                     bool rvalue = rhs.GetBoolean(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::INT8:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     int8_t lvalue = lhs.GetByte(field_idx);
                     int8_t rvalue = rhs.GetByte(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::INT16:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     int16_t lvalue = lhs.GetShort(field_idx);
                     int16_t rvalue = rhs.GetShort(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::DATE32:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     int32_t lvalue = lhs.GetDate(field_idx);
                     int32_t rvalue = rhs.GetDate(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
 
         case arrow::Type::type::INT32:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     int32_t lvalue = lhs.GetInt(field_idx);
                     int32_t rvalue = rhs.GetInt(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::INT64:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     int64_t lvalue = lhs.GetLong(field_idx);
                     int64_t rvalue = rhs.GetLong(field_idx);
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::FLOAT:
-            // TODO(xinyu.lxy):
-            // currently in java KeyComparatorSupplier: -inf < -0.0 == +0.0 < +inf = nan
-            // paimon-cpp: -inf < -0.0 == +0.0 < +inf and nan cannot be compared
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     float lvalue = lhs.GetFloat(field_idx);
                     float rvalue = rhs.GetFloat(field_idx);
-                    return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
+                    return CompareFloatingPoint(lvalue, rvalue);
                 });
         case arrow::Type::type::DOUBLE:
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     double lvalue = lhs.GetDouble(field_idx);
                     double rvalue = rhs.GetDouble(field_idx);
-                    return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
+                    return CompareFloatingPoint(lvalue, rvalue);
                 });
         case arrow::Type::type::STRING:
         case arrow::Type::type::BINARY: {
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     auto lvalue = lhs.GetStringView(field_idx);
                     auto rvalue = rhs.GetStringView(field_idx);
@@ -157,7 +154,7 @@ Result<FieldsComparator::FieldComparatorFunc> FieldsComparator::CompareField(
         case arrow::Type::type::TIMESTAMP: {
             auto timestamp_type = checked_pointer_cast<arrow::TimestampType>(input_type);
             int32_t precision = DateTimeUtils::GetPrecisionFromType(timestamp_type);
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx, precision](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
                     Timestamp lvalue = lhs.GetTimestamp(field_idx, precision);
                     Timestamp rvalue = rhs.GetTimestamp(field_idx, precision);
@@ -168,7 +165,7 @@ Result<FieldsComparator::FieldComparatorFunc> FieldsComparator::CompareField(
             auto* decimal_type = checked_cast<arrow::Decimal128Type*>(input_type.get());
             auto precision = decimal_type->precision();
             auto scale = decimal_type->scale();
-            return FieldsComparator::FieldComparatorFunc(
+            return FieldComparatorFunc(
                 [field_idx, precision, scale](const InternalRow& lhs,
                                               const InternalRow& rhs) -> int32_t {
                     Decimal lvalue = lhs.GetDecimal(field_idx, precision, scale);

src/paimon/common/utils/fields_comparator.h

@@ -41,6 +41,9 @@ class DataField;
 /// A `Comparator` that compares the file store key.
 class FieldsComparator {
  public:
+    using FieldComparatorFunc =
+        std::function<int32_t(const InternalRow& lhs, const InternalRow& rhs)>;
+
     static Result<std::unique_ptr<FieldsComparator>> Create(
         const std::vector<DataField>& input_data_field, bool is_ascending_order);
 
@@ -82,9 +85,6 @@ class FieldsComparator {
     }
 
  private:
-    using FieldComparatorFunc =
-        std::function<int32_t(const InternalRow& lhs, const InternalRow& rhs)>;
-
     FieldsComparator(bool is_ascending_order, const std::vector<int32_t>& sort_fields,
                      std::vector<FieldComparatorFunc>&& comparators)
         : is_ascending_order_(is_ascending_order),

src/paimon/common/utils/fields_comparator_test.cpp

@@ -20,8 +20,10 @@
 #include "paimon/common/utils/fields_comparator.h"
 
 #include <cstddef>
+#include <limits>
 #include <string>
 #include <variant>
+#include <vector>
 
 #include "arrow/api.h"
 #include "gtest/gtest.h"
@@ -80,6 +82,44 @@ class FieldsComparatorTest : public ::testing::Test {
         }
         CheckResult(row1, row2, input_types, sort_fields, has_null);
     }
+
+    template <typename T>
+    void CheckFloatingPointOrder(const std::shared_ptr<arrow::DataType>& type) {
+        std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+        std::vector<DataField> data_fields = {DataField(0, arrow::field("f0", type))};
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldsComparator> ascending_comparator,
+                             FieldsComparator::Create(data_fields,
+                                                      /*is_ascending_order=*/true));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldsComparator> descending_comparator,
+                             FieldsComparator::Create(data_fields,
+                                                      /*is_ascending_order=*/false));
+
+        const T nan = std::numeric_limits<T>::quiet_NaN();
+        const std::vector<T> values = {-std::numeric_limits<T>::infinity(),
+                                       static_cast<T>(-1),
+                                       static_cast<T>(-0.0),
+                                       static_cast<T>(0.0),
+                                       static_cast<T>(1),
+                                       std::numeric_limits<T>::infinity(),
+                                       nan};
+        std::vector<BinaryRow> rows;
+        rows.reserve(values.size());
+        for (T value : values) {
+            rows.emplace_back(BinaryRowGenerator::GenerateRow({value}, pool.get()));
+        }
+
+        for (size_t i = 0; i < rows.size(); ++i) {
+            for (size_t j = 0; j < rows.size(); ++j) {
+                int32_t expected = i == j ? 0 : (i < j ? -1 : 1);
+                ASSERT_EQ(expected, ascending_comparator->CompareTo(rows[i], rows[j]));
+                ASSERT_EQ(-expected, descending_comparator->CompareTo(rows[i], rows[j]));
+            }
+        }
+
+        BinaryRow negative_nan_row = BinaryRowGenerator::GenerateRow({-nan}, pool.get());
+        ASSERT_EQ(0, ascending_comparator->CompareTo(rows.back(), negative_nan_row));
+        ASSERT_EQ(0, ascending_comparator->CompareTo(negative_nan_row, rows.back()));
+    }
 };
 
 TEST_F(FieldsComparatorTest, TestSimple) {
@@ -202,6 +242,11 @@ TEST_F(FieldsComparatorTest, TestSimple) {
     }
 }
 
+TEST_F(FieldsComparatorTest, TestFloatingPointOrder) {
+    CheckFloatingPointOrder<float>(arrow::float32());
+    CheckFloatingPointOrder<double>(arrow::float64());
+}
+
 TEST_F(FieldsComparatorTest, TestTimestampType) {
     auto pool = GetDefaultPool();
     // test ts with different precision