[core] Support reading sequence_number in AuditLogTable and BinlogTable #6858
Conversation
wzhero1
force-pushed
the
auditlog-support-seqnum
branch
9 times, most recently
from
ed453a8 to
eae1a70
Compare
wzhero1
force-pushed
the
auditlog-support-seqnum
branch
from
eae1a70 to
7ab955a
Compare
yunfengzhou-hub
left a comment
yunfengzhou-hub
left a comment
There was a problem hiding this comment.
Thanks for the PR! Left some comments as below.
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
| @Nullable int[] projection) { | ||
| if (systemFieldExtractors.isEmpty()) { | ||
| // No system fields, use the default unwrap logic | ||
| return KeyValueTableRead.unwrap(reader); |
There was a problem hiding this comment.
It might be better to merge this method into KeyValueTableRead#unwrap. As a decorator/wrapper to this method, extending the original method might be more beneficial to extensibility.
There was a problem hiding this comment.
Done. I've moved the logic into KeyValueTableRead.java. The method structure is now:
unwrap(RecordReader<KeyValue> reader, List<SpecialFieldExtractor> specialFieldExtractors, @Nullable int[] projection)as public method for extensibilityunwrap(RecordReader<KeyValue> reader)as private method for internal use
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
| checkAnswer( | ||
| sql("SELECT * FROM paimon_incremental_query('`t$audit_log`', 'tag1', 'tag2') ORDER BY id"), | ||
| Seq(Row("-D", 999))) | ||
| Seq(Row("-D", 100002L, 999))) |
There was a problem hiding this comment.
100002L is the value of id column, rather than sequence number. Not sure whether it is correct here.
There was a problem hiding this comment.
999 is the value of id, 100002L is the sequence number
| } | ||
|
|
||
| /** Creates a table with changelog producer enabled. */ | ||
| private FileStoreTable createChangelogTable(String tableName) throws Exception { |
There was a problem hiding this comment.
This method is only used in one place. So compared with introducing this method, it might be better to preserve the original code structure and commit history.
There was a problem hiding this comment.
While it's currently used in one place, I prefer keeping this helper method for single responsibility and better extensibility. It encapsulates the changelog table creation logic cleanly, making future test additions easier without duplicating configuration code.
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
| */ | ||
| @Nullable | ||
| public static SystemFieldExtractor getExtractor(String fieldName) { | ||
| return EXTRACTOR_REGISTRY.get(fieldName); |
There was a problem hiding this comment.
This method only checks field names. It might be better to also check field data types.
There was a problem hiding this comment.
For special fields (like _ROWKIND, _SEQUENCE_NUMBER), field name matching is sufficient.
paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueSystemFieldsRecordReader.java
Outdated
Show resolved
Hide resolved
2 participants
Purpose
This PR adds support for accessing system fields (specifically
sequence_number) inAuditLogTableandBinlogTable.Currently, downstream consumers often need the
sequence_numberto ensure strict ordering or to implement exactly-once processing when reading from audit logs or binlogs. However, these system tables previously did not expose this field.This change implements
KeyValueSystemFieldsRecordReaderto enable the projection of system fields for KeyValue records, allowing users to querysequence_numberdirectly from audit log and binlog tables.Tests
IncrementalReadSystemFieldsTestto verify functionality.AuditLogTableTestandBinlogTableTestto coversequence_numberread cases.API and Format
Documentation
docs/content/concepts/system-tables.mdneeds to change