feat: add audit_log and binlog system tables (#268)

Author: suxiaogang223

cmake_modules/SetupCxxFlags.cmake

@@ -81,12 +81,10 @@ if("${BUILD_WARNING_LEVEL}" STREQUAL "CHECKIN")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unknown-warning-option")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-constant-logical-operand")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-declarations")
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-builtins")
     elseif(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-conversion")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-declarations")
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-builtins")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-sign-conversion")
         set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-unused-variable")
     else()
@@ -192,7 +190,6 @@ elseif(CMAKE_CXX_COMPILER_ID STREQUAL "AppleClang" OR CMAKE_CXX_COMPILER_ID STRE
     # Don't complain about optimization passes that were not possible
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-pass-failed")
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-declarations")
-    set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated-builtins")
 
     if(CMAKE_CXX_COMPILER_ID STREQUAL "AppleClang")
         # Depending on the default OSX_DEPLOYMENT_TARGET (< 10.9), libstdc++ may be

include/paimon/defs.h

@@ -374,6 +374,14 @@ struct PAIMON_EXPORT Options {
     /// "aggregation.remove-record-on-delete" - Whether to remove the whole row in aggregation
     /// engine when delete records are received. Default value is "false".
     static const char AGGREGATION_REMOVE_RECORD_ON_DELETE[];
+    /// "table-read.sequence-number.enabled" - Whether to include the _SEQUENCE_NUMBER field when
+    /// reading the audit_log or binlog system tables. This is only valid for primary key tables.
+    /// Default value is "false".
+    static const char TABLE_READ_SEQUENCE_NUMBER_ENABLED[];
+    /// "key-value.sequence-number.enabled" - Whether to include the _SEQUENCE_NUMBER field when
+    /// reading key-value data. This is an internal option used by AuditLogTable and BinlogTable
+    /// when table-read.sequence-number.enabled is set to true. Default value is "false".
+    static const char KEY_VALUE_SEQUENCE_NUMBER_ENABLED[];
 
     /// "scan.timestamp-millis" - Optional timestamp used in case of "from-timestamp" scan mode.
     /// For batch sources, produces the latest snapshot earlier than or equal to the timestamp.

include/paimon/utils/special_field_ids.h

@@ -25,18 +25,20 @@ namespace paimon {
 class SpecialFieldIds {
  protected:
     /// System defined constant for field id boundary. Value: INT32_MAX - 10000
-    static const int32_t CPP_FIELD_ID_END;
+    static const int32_t CPP_FIELD_ID_END = std::numeric_limits<int32_t>::max() - 10000;
 
  public:
     /// Special field ID reserved for sequence number. Value: INT32_MAX - 1
-    static const int32_t SEQUENCE_NUMBER;
+    static const int32_t SEQUENCE_NUMBER = std::numeric_limits<int32_t>::max() - 1;
     /// Special field ID reserved for value kind. Value: INT32_MAX - 2
-    static const int32_t VALUE_KIND;
+    static const int32_t VALUE_KIND = std::numeric_limits<int32_t>::max() - 2;
+    /// Special field ID reserved for row kind. Value: INT32_MAX - 3
+    static const int32_t ROW_KIND = std::numeric_limits<int32_t>::max() - 3;
     /// Special field ID reserved for row ID. Value: INT32_MAX - 5
-    static const int32_t ROW_ID;
+    static const int32_t ROW_ID = std::numeric_limits<int32_t>::max() - 5;
 
     /// Special field ID reserved for index score. Value: CPP_FIELD_ID_END - 1
-    static const int32_t INDEX_SCORE;
+    static const int32_t INDEX_SCORE = CPP_FIELD_ID_END - 1;
 };
 
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -313,9 +313,10 @@ set(PAIMON_CORE_SRCS
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
     core/table/source/data_evolution_batch_scan.cpp
+    core/table/system/audit_log_system_table.cpp
+    core/table/system/binlog_system_table.cpp
     core/table/system/options_system_table.cpp
     core/table/system/system_table.cpp
-    core/table/system/system_table_read.cpp
     core/table/system/system_table_scan.cpp
     core/table/system/system_table_schema.cpp
     core/tag/tag.cpp
@@ -695,6 +696,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/split_generator_test.cpp
                     core/table/source/startup_mode_test.cpp
                     core/table/source/table_scan_test.cpp
+                    core/table/system/system_table_test.cpp
                     core/tag/tag_test.cpp
                     core/utils/branch_manager_test.cpp
                     core/utils/file_store_path_factory_cache_test.cpp

src/paimon/common/defs.cpp

@@ -94,6 +94,8 @@ const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 const char Options::GLOBAL_INDEX_THREAD_NUM[] = "global-index.thread-num";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
 const char Options::AGGREGATION_REMOVE_RECORD_ON_DELETE[] = "aggregation.remove-record-on-delete";
+const char Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED[] = "table-read.sequence-number.enabled";
+const char Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED[] = "key-value.sequence-number.enabled";
 const char Options::SCAN_TIMESTAMP_MILLIS[] = "scan.timestamp-millis";
 const char Options::SCAN_TIMESTAMP[] = "scan.timestamp";
 const char Options::SCAN_TAG_NAME[] = "scan.tag-name";

src/paimon/common/table/special_fields.h

@@ -45,6 +45,12 @@ struct SpecialFields {
         return data_field;
     }
 
+    static const DataField& RowKind() {
+        static const DataField data_field =
+            DataField(SpecialFieldIds::ROW_KIND, arrow::field("rowkind", arrow::utf8()));
+        return data_field;
+    }
+
     static const DataField& RowId() {
         static const DataField data_field =
             DataField(SpecialFieldIds::ROW_ID, arrow::field("_ROW_ID", arrow::int64()));

src/paimon/common/table/special_fields_test.cpp

@@ -35,6 +35,12 @@ TEST(SpecialFieldsTest, TestValueKindField) {
     ASSERT_EQ(SpecialFields::ValueKind().Type()->id(), arrow::Type::INT8);
 }
 
+TEST(SpecialFieldsTest, TestRowKindField) {
+    ASSERT_EQ(SpecialFields::RowKind().Id(), std::numeric_limits<int32_t>::max() - 3);
+    ASSERT_EQ(SpecialFields::RowKind().Name(), "rowkind");
+    ASSERT_EQ(SpecialFields::RowKind().Type()->id(), arrow::Type::STRING);
+}
+
 TEST(SpecialFieldsTest, TestRowIdField) {
     ASSERT_EQ(SpecialFields::RowId().Id(), std::numeric_limits<int32_t>::max() - 5);
     ASSERT_EQ(SpecialFields::RowId().Name(), "_ROW_ID");

src/paimon/core/catalog/file_system_catalog.cpp

@@ -272,11 +272,18 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
         if (!latest_schema) {
             return Status::NotExist(fmt::format("{} not exist", data_identifier.ToString()));
         }
-        PAIMON_ASSIGN_OR_RAISE(
-            std::shared_ptr<SystemTable> system_table,
-            SystemTableLoader::Load(system_table_name.value(), fs_, GetTableLocation(identifier),
-                                    latest_schema.value()));
-        return std::make_shared<SystemTableSchema>(system_table->ArrowSchema());
+        std::map<std::string, std::string> dynamic_options;
+        PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName());
+        if (branch) {
+            dynamic_options[Options::BRANCH] = branch.value();
+        }
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<SystemTable> system_table,
+                               SystemTableLoader::Load(system_table_name.value(), fs_,
+                                                       GetTableLocation(data_identifier),
+                                                       latest_schema.value(), dynamic_options));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> arrow_schema,
+                               system_table->ArrowSchema());
+        return std::make_shared<SystemTableSchema>(std::move(arrow_schema));
     }
     PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
                            TableSchemaExists(identifier));

src/paimon/core/catalog/file_system_catalog_test.cpp

@@ -210,6 +210,77 @@ TEST(FileSystemCatalogTest, TestOptionsSystemTableCatalog) {
                         "Cannot rename system table");
 }
 
+TEST(FileSystemCatalogTest, TestAuditLogAndBinlogSystemTableCatalog) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str());
+    ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true));
+
+    auto typed_schema =
+        arrow::schema({arrow::field("pk", arrow::utf8()), arrow::field("v", arrow::int32(), true)});
+    ::ArrowSchema schema;
+    ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok());
+    ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema,
+                                  /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, options,
+                                  /*ignore_if_exists=*/false));
+    ArrowSchemaRelease(&schema);
+
+    Identifier audit_log_identifier("db1", "tbl1$audit_log");
+    Identifier binlog_identifier("db1", "tbl1$binlog");
+    ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(audit_log_identifier));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(binlog_identifier));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("db1", "tbl1$unknown")));
+    ASSERT_FALSE(exists);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> audit_log_system_schema,
+                         catalog.LoadTableSchema(audit_log_identifier));
+    ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(audit_log_system_schema) != nullptr);
+    ASSERT_OK_AND_ASSIGN(auto audit_log_c_schema, audit_log_system_schema->GetArrowSchema());
+    auto audit_log_schema_result = arrow::ImportSchema(audit_log_c_schema.get());
+    ASSERT_TRUE(audit_log_schema_result.ok()) << audit_log_schema_result.status().ToString();
+    auto audit_log_schema = audit_log_schema_result.ValueUnsafe();
+    ASSERT_EQ(audit_log_schema->field_names(), (std::vector<std::string>{"rowkind", "pk", "v"}));
+    ASSERT_EQ(audit_log_schema->field(0)->type()->id(), arrow::Type::STRING);
+    ASSERT_EQ(audit_log_schema->field(1)->type()->id(), arrow::Type::STRING);
+    ASSERT_EQ(audit_log_schema->field(2)->type()->id(), arrow::Type::INT32);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> binlog_system_schema,
+                         catalog.LoadTableSchema(binlog_identifier));
+    ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(binlog_system_schema) != nullptr);
+    ASSERT_OK_AND_ASSIGN(auto binlog_c_schema, binlog_system_schema->GetArrowSchema());
+    auto binlog_schema_result = arrow::ImportSchema(binlog_c_schema.get());
+    ASSERT_TRUE(binlog_schema_result.ok()) << binlog_schema_result.status().ToString();
+    auto binlog_schema = binlog_schema_result.ValueUnsafe();
+    ASSERT_EQ(binlog_schema->field_names(), (std::vector<std::string>{"rowkind", "pk", "v"}));
+    ASSERT_EQ(binlog_schema->field(0)->type()->id(), arrow::Type::STRING);
+    ASSERT_EQ(binlog_schema->field(1)->type()->id(), arrow::Type::LIST);
+    ASSERT_EQ(binlog_schema->field(2)->type()->id(), arrow::Type::LIST);
+    auto binlog_pk_type =
+        std::dynamic_pointer_cast<arrow::ListType>(binlog_schema->field(1)->type());
+    auto binlog_v_type =
+        std::dynamic_pointer_cast<arrow::ListType>(binlog_schema->field(2)->type());
+    ASSERT_TRUE(binlog_pk_type);
+    ASSERT_TRUE(binlog_v_type);
+    ASSERT_EQ(binlog_pk_type->value_type()->id(), arrow::Type::STRING);
+    ASSERT_EQ(binlog_v_type->value_type()->id(), arrow::Type::INT32);
+
+    ::ArrowSchema system_create_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok());
+    ASSERT_NOK_WITH_MSG(
+        catalog.CreateTable(audit_log_identifier, &system_create_schema, {}, {}, options, false),
+        "Cannot create table for system table");
+    ArrowSchemaRelease(&system_create_schema);
+    ASSERT_NOK_WITH_MSG(catalog.DropTable(binlog_identifier, false), "Cannot drop system table");
+    ASSERT_NOK_WITH_MSG(catalog.RenameTable(audit_log_identifier, Identifier("db1", "tbl2"), false),
+                        "Cannot rename system table");
+}
+
 TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
     std::map<std::string, std::string> options;
     options[Options::FILE_SYSTEM] = "local";

src/paimon/core/core_options.cpp

@@ -424,6 +424,8 @@ struct CoreOptions::Impl {
     bool lookup_wait = true;
     bool partial_update_remove_record_on_delete = false;
     bool aggregation_remove_record_on_delete = false;
+    bool table_read_sequence_number_enabled = false;
+    bool key_value_sequence_number_enabled = false;
     bool file_index_read_enabled = true;
     bool enable_adaptive_prefetch_strategy = true;
     bool index_file_in_data_file_dir = false;
@@ -629,6 +631,12 @@ struct CoreOptions::Impl {
         // Parse aggregation_remove_record_on_delete
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::AGGREGATION_REMOVE_RECORD_ON_DELETE,
                                                 &aggregation_remove_record_on_delete));
+        // Parse table-read.sequence-number.enabled - expose sequence number in system tables
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED,
+                                                &table_read_sequence_number_enabled));
+        // Parse key-value.sequence-number.enabled - internal sequence number read switch
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED,
+                                                &key_value_sequence_number_enabled));
         // Parse partial-update.remove-record-on-sequence-group
         PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
             Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP, Options::FIELDS_SEPARATOR,
@@ -1201,6 +1209,14 @@ bool CoreOptions::AggregationRemoveRecordOnDelete() const {
     return impl_->aggregation_remove_record_on_delete;
 }
 
+bool CoreOptions::TableReadSequenceNumberEnabled() const {
+    return impl_->table_read_sequence_number_enabled;
+}
+
+bool CoreOptions::KeyValueSequenceNumberEnabled() const {
+    return impl_->key_value_sequence_number_enabled;
+}
+
 std::vector<std::string> CoreOptions::GetPartialUpdateRemoveRecordOnSequenceGroup() const {
     return impl_->remove_record_on_sequence_group;
 }