This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 4286a058 [Improve](source) Follow the Doris rename of row binlog to
changelog (#710)
4286a058 is described below
commit 4286a0584ccd70b5b86129ff4fe995f5268e7daf
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Oct 21 10:48:35 2026 +0100
[Improve](source) Follow the Doris rename of row binlog to changelog (#710)
Doris renamed the unreleased row binlog feature, which the incremental
source reads, to changelog (apache/doris#69311). A table now enables it
with "changelog.enable" = "true" instead of "binlog.enable" = "true"
plus "binlog.format" = "ROW", Doris rejects the old binlog.format and
binlog.need_historical_value properties, and the hidden columns are now
__DORIS_CHANGELOG_TSO__, __DORIS_CHANGELOG_LSN__ and
__DORIS_CHANGELOG_OP__. The @incr syntax, the increment types,
information_schema.tso_status and error code 5101 are unchanged. The
source still selected and ordered by the __DORIS_BINLOG_*__ columns, so
it could not read Doris master after that rename.
Query and read the new hidden columns, and rename the connector's own
row binlog names to changelog so they match the Doris feature:
- source.binlog.increment-type -> source.changelog.increment-type
- source.binlog.poll-interval -> source.changelog.poll-interval
- source.binlog.visible-wait-timeout ->
source.changelog.visible-wait-timeout
- source.binlog.offset-table -> source.changelog.offset-table
- source.binlog.consumer-id -> source.changelog.consumer-id
- DorisConfigOptions.SOURCE_BINLOG_* -> SOURCE_CHANGELOG_*
- ConfigurationOptions.SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT
-> SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT
- DorisBinlogIncrementType -> DorisChangelogIncrementType
- DorisReadOptions getBinlog*() and Builder.setBinlog*()
-> getChangelog*() and setChangelog*()
- DorisSourceRecord getBinlogTso/Lsn/Op() -> getChangelogTso/Lsn/Op()
This is a hard rename without deprecated keys: a Flink table that still
sets a source.binlog.* option fails option validation as an unsupported
option. Those keys only ever worked against Doris master builds from
before the rename, and the connector no longer reads such builds. The
integration and end-to-end fixtures create their tables with
"changelog.enable" and "changelog.need_historical_value". MySQL binlog,
Flink CDC and Debezium code is unchanged, and so are the split and
checkpoint formats.
Co-authored-by: morningman <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../doris/flink/cfg/ConfigurationOptions.java | 2 +-
.../apache/doris/flink/cfg/DorisReadOptions.java | 138 +++++++++++----------
.../org/apache/doris/flink/rest/RestService.java | 4 +-
.../apache/doris/flink/serialization/RowBatch.java | 50 ++++----
...tType.java => DorisChangelogIncrementType.java} | 10 +-
.../source/assigners/DorisSourceSplitAssigner.java | 10 +-
.../source/reader/DorisFlightValueReader.java | 12 +-
.../flink/source/reader/DorisSourceRecord.java | 36 +++---
.../source/reader/DorisSourceSplitReader.java | 6 +-
.../doris/flink/source/split/DorisStreamSplit.java | 2 +-
.../doris/flink/table/DorisConfigOptions.java | 29 ++---
.../DorisIncrementalRowBatchTest.java | 16 +--
.../assigners/DorisSourceOffsetRestoreTest.java | 4 +-
.../DorisHybridSourceEnumeratorTest.java | 6 +-
.../enumerator/DorisSourceEnumeratorTest.java | 2 +-
.../source/reader/DorisFlightValueReaderTest.java | 6 +-
.../source/reader/DorisRecordEmitterTest.java | 2 +-
...test_doris_incremental_all_types_source_tbl.sql | 5 +-
.../org/apache/doris/flink/source/DorisSource.java | 26 ++--
.../flink/source/reader/DorisSourceReader.java | 2 +-
.../flink/table/DorisDynamicTableFactory.java | 43 +++----
.../doris/flink/table/DorisDynamicTableSource.java | 4 +-
.../doris/flink/source/DorisSourceOptionsTest.java | 67 +++++-----
.../flink/table/DorisDynamicTableFactoryTest.java | 14 +--
.../flink/table/DorisDynamicTableSourceTest.java | 12 +-
.../org/apache/doris/flink/source/DorisSource.java | 26 ++--
.../flink/source/reader/DorisSourceReader.java | 2 +-
.../flink/table/DorisDynamicTableFactory.java | 43 +++----
.../doris/flink/table/DorisDynamicTableSource.java | 4 +-
.../flink/table/DorisDynamicTableFactoryTest.java | 14 +--
.../flink/table/DorisDynamicTableSourceTest.java | 12 +-
.../e2e/DorisIncrementalSourceE2ECase.java | 4 +-
.../source/AbstractDorisIncrementalITCase.java | 17 ++-
.../flink/source/DorisIncrementalSourceITCase.java | 2 +-
34 files changed, 324 insertions(+), 308 deletions(-)
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
index 5a77240a..22465095 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
@@ -64,7 +64,7 @@ public interface ConfigurationOptions {
String FLIGHT_SQL_PORT = "source.flight-sql-port";
Integer FLIGHT_SQL_PORT_DEFAULT = -1;
- Long SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT = 5 * 60 * 1000L;
+ Long SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT = 5 * 60 * 1000L;
String SINK_HTTP_UTF8_CHARSET = "sink.http-utf8-charset";
Boolean SINK_HTTP_UTF8_CHARSET_DEFAULT = false;
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
index 6d03e73d..36edc941 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
@@ -17,7 +17,7 @@
package org.apache.doris.flink.cfg;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSourceScanMode;
import java.io.Serializable;
@@ -47,11 +47,11 @@ public class DorisReadOptions implements Serializable {
private Integer thriftMaxMessageSize;
private DorisSourceScanMode scanMode;
private String scanTimestamp;
- private DorisBinlogIncrementType binlogIncrementType;
- private Long binlogPollIntervalMs;
- private Long binlogVisibleWaitTimeoutMs;
- private String binlogOffsetTable;
- private String binlogConsumerId;
+ private DorisChangelogIncrementType changelogIncrementType;
+ private Long changelogPollIntervalMs;
+ private Long changelogVisibleWaitTimeoutMs;
+ private String changelogOffsetTable;
+ private String changelogConsumerId;
public DorisReadOptions(
String readFields,
@@ -72,9 +72,9 @@ public class DorisReadOptions implements Serializable {
Integer thriftMaxMessageSize,
DorisSourceScanMode scanMode,
String scanTimestamp,
- DorisBinlogIncrementType binlogIncrementType,
- Long binlogPollIntervalMs,
- Long binlogVisibleWaitTimeoutMs) {
+ DorisChangelogIncrementType changelogIncrementType,
+ Long changelogPollIntervalMs,
+ Long changelogVisibleWaitTimeoutMs) {
this(
readFields,
filterQuery,
@@ -94,9 +94,9 @@ public class DorisReadOptions implements Serializable {
thriftMaxMessageSize,
scanMode,
scanTimestamp,
- binlogIncrementType,
- binlogPollIntervalMs,
- binlogVisibleWaitTimeoutMs,
+ changelogIncrementType,
+ changelogPollIntervalMs,
+ changelogVisibleWaitTimeoutMs,
null,
null);
}
@@ -120,11 +120,11 @@ public class DorisReadOptions implements Serializable {
Integer thriftMaxMessageSize,
DorisSourceScanMode scanMode,
String scanTimestamp,
- DorisBinlogIncrementType binlogIncrementType,
- Long binlogPollIntervalMs,
- Long binlogVisibleWaitTimeoutMs,
- String binlogOffsetTable,
- String binlogConsumerId) {
+ DorisChangelogIncrementType changelogIncrementType,
+ Long changelogPollIntervalMs,
+ Long changelogVisibleWaitTimeoutMs,
+ String changelogOffsetTable,
+ String changelogConsumerId) {
this.readFields = readFields;
this.filterQuery = filterQuery;
this.requestTabletSize = requestTabletSize;
@@ -143,11 +143,11 @@ public class DorisReadOptions implements Serializable {
this.thriftMaxMessageSize = thriftMaxMessageSize;
this.scanMode = scanMode;
this.scanTimestamp = scanTimestamp;
- this.binlogIncrementType = binlogIncrementType;
- this.binlogPollIntervalMs = binlogPollIntervalMs;
- this.binlogVisibleWaitTimeoutMs = binlogVisibleWaitTimeoutMs;
- this.binlogOffsetTable = binlogOffsetTable;
- this.binlogConsumerId = binlogConsumerId;
+ this.changelogIncrementType = changelogIncrementType;
+ this.changelogPollIntervalMs = changelogPollIntervalMs;
+ this.changelogVisibleWaitTimeoutMs = changelogVisibleWaitTimeoutMs;
+ this.changelogOffsetTable = changelogOffsetTable;
+ this.changelogConsumerId = changelogConsumerId;
}
public String getReadFields() {
@@ -238,24 +238,24 @@ public class DorisReadOptions implements Serializable {
return scanTimestamp;
}
- public DorisBinlogIncrementType getBinlogIncrementType() {
- return binlogIncrementType;
+ public DorisChangelogIncrementType getChangelogIncrementType() {
+ return changelogIncrementType;
}
- public Long getBinlogPollIntervalMs() {
- return binlogPollIntervalMs;
+ public Long getChangelogPollIntervalMs() {
+ return changelogPollIntervalMs;
}
- public Long getBinlogVisibleWaitTimeoutMs() {
- return binlogVisibleWaitTimeoutMs;
+ public Long getChangelogVisibleWaitTimeoutMs() {
+ return changelogVisibleWaitTimeoutMs;
}
- public String getBinlogOffsetTable() {
- return binlogOffsetTable;
+ public String getChangelogOffsetTable() {
+ return changelogOffsetTable;
}
- public String getBinlogConsumerId() {
- return binlogConsumerId;
+ public String getChangelogConsumerId() {
+ return changelogConsumerId;
}
public static Builder builder() {
@@ -293,11 +293,11 @@ public class DorisReadOptions implements Serializable {
&& Objects.equals(thriftMaxMessageSize,
that.thriftMaxMessageSize)
&& scanMode == that.scanMode
&& Objects.equals(scanTimestamp, that.scanTimestamp)
- && binlogIncrementType == that.binlogIncrementType
- && Objects.equals(binlogPollIntervalMs,
that.binlogPollIntervalMs)
- && Objects.equals(binlogVisibleWaitTimeoutMs,
that.binlogVisibleWaitTimeoutMs)
- && Objects.equals(binlogOffsetTable, that.binlogOffsetTable)
- && Objects.equals(binlogConsumerId, that.binlogConsumerId);
+ && changelogIncrementType == that.changelogIncrementType
+ && Objects.equals(changelogPollIntervalMs,
that.changelogPollIntervalMs)
+ && Objects.equals(changelogVisibleWaitTimeoutMs,
that.changelogVisibleWaitTimeoutMs)
+ && Objects.equals(changelogOffsetTable,
that.changelogOffsetTable)
+ && Objects.equals(changelogConsumerId,
that.changelogConsumerId);
}
@Override
@@ -321,11 +321,11 @@ public class DorisReadOptions implements Serializable {
thriftMaxMessageSize,
scanMode,
scanTimestamp,
- binlogIncrementType,
- binlogPollIntervalMs,
- binlogVisibleWaitTimeoutMs,
- binlogOffsetTable,
- binlogConsumerId);
+ changelogIncrementType,
+ changelogPollIntervalMs,
+ changelogVisibleWaitTimeoutMs,
+ changelogOffsetTable,
+ changelogConsumerId);
}
public DorisReadOptions copy() {
@@ -348,11 +348,11 @@ public class DorisReadOptions implements Serializable {
thriftMaxMessageSize,
scanMode,
scanTimestamp,
- binlogIncrementType,
- binlogPollIntervalMs,
- binlogVisibleWaitTimeoutMs,
- binlogOffsetTable,
- binlogConsumerId);
+ changelogIncrementType,
+ changelogPollIntervalMs,
+ changelogVisibleWaitTimeoutMs,
+ changelogOffsetTable,
+ changelogConsumerId);
}
/** Builder of {@link DorisReadOptions}. */
@@ -382,12 +382,13 @@ public class DorisReadOptions implements Serializable {
ConfigurationOptions.DORIS_THRIFT_MAX_MESSAGE_SIZE_DEFAULT;
private DorisSourceScanMode scanMode = DorisSourceScanMode.SNAPSHOT;
private String scanTimestamp;
- private DorisBinlogIncrementType binlogIncrementType =
DorisBinlogIncrementType.DETAIL;
- private Long binlogPollIntervalMs = 10_000L;
- private Long binlogVisibleWaitTimeoutMs =
-
ConfigurationOptions.SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT;
- private String binlogOffsetTable;
- private String binlogConsumerId;
+ private DorisChangelogIncrementType changelogIncrementType =
+ DorisChangelogIncrementType.DETAIL;
+ private Long changelogPollIntervalMs = 10_000L;
+ private Long changelogVisibleWaitTimeoutMs =
+
ConfigurationOptions.SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT;
+ private String changelogOffsetTable;
+ private String changelogConsumerId;
/**
* Sets the readFields for doris table to push down projection, such
as name,age.
@@ -568,28 +569,29 @@ public class DorisReadOptions implements Serializable {
return this;
}
- public Builder setBinlogIncrementType(DorisBinlogIncrementType
binlogIncrementType) {
- this.binlogIncrementType = binlogIncrementType;
+ public Builder setChangelogIncrementType(
+ DorisChangelogIncrementType changelogIncrementType) {
+ this.changelogIncrementType = changelogIncrementType;
return this;
}
- public Builder setBinlogPollIntervalMs(Long binlogPollIntervalMs) {
- this.binlogPollIntervalMs = binlogPollIntervalMs;
+ public Builder setChangelogPollIntervalMs(Long
changelogPollIntervalMs) {
+ this.changelogPollIntervalMs = changelogPollIntervalMs;
return this;
}
- public Builder setBinlogVisibleWaitTimeoutMs(Long
binlogVisibleWaitTimeoutMs) {
- this.binlogVisibleWaitTimeoutMs = binlogVisibleWaitTimeoutMs;
+ public Builder setChangelogVisibleWaitTimeoutMs(Long
changelogVisibleWaitTimeoutMs) {
+ this.changelogVisibleWaitTimeoutMs = changelogVisibleWaitTimeoutMs;
return this;
}
- public Builder setBinlogOffsetTable(String binlogOffsetTable) {
- this.binlogOffsetTable = binlogOffsetTable;
+ public Builder setChangelogOffsetTable(String changelogOffsetTable) {
+ this.changelogOffsetTable = changelogOffsetTable;
return this;
}
- public Builder setBinlogConsumerId(String binlogConsumerId) {
- this.binlogConsumerId = binlogConsumerId;
+ public Builder setChangelogConsumerId(String changelogConsumerId) {
+ this.changelogConsumerId = changelogConsumerId;
return this;
}
@@ -618,11 +620,11 @@ public class DorisReadOptions implements Serializable {
thriftMaxMessageSize,
scanMode,
scanTimestamp,
- binlogIncrementType,
- binlogPollIntervalMs,
- binlogVisibleWaitTimeoutMs,
- binlogOffsetTable,
- binlogConsumerId);
+ changelogIncrementType,
+ changelogPollIntervalMs,
+ changelogVisibleWaitTimeoutMs,
+ changelogOffsetTable,
+ changelogConsumerId);
}
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
index f93f2899..16bdd357 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/rest/RestService.java
@@ -104,7 +104,9 @@ public class RestService implements Serializable {
"SELECT FROM_UNIXTIME(CURRENT_TSO_PHYSICAL_TIME / 1000, "
+ "'%Y-%m-%d %H:%i:%s') FROM
information_schema.tso_status";
- /** Resolves the current Doris TSO to the timestamp format accepted by
row-binlog queries. */
+ /**
+ * Resolves the current Doris TSO to the timestamp format accepted by
Doris changelog queries.
+ */
public static String resolveCurrentTimestamp(
DorisOptions options, DorisReadOptions readOptions, Logger logger)
{
List<String> endpoints = allEndpoints(options.getFenodes(), logger);
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
index 47f357de..24236099 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
@@ -84,9 +84,9 @@ public class RowBatch {
public static class Row {
private final List<Object> cols;
- private Long binlogTso;
- private Long binlogLsn;
- private Long binlogOp;
+ private Long changelogTso;
+ private Long changelogLsn;
+ private Long changelogOp;
Row(int colCount) {
this.cols = new ArrayList<>(colCount);
@@ -102,7 +102,7 @@ public class RowBatch {
private DorisSourceRecord toSourceRecord(boolean incremental) {
return incremental
- ? DorisSourceRecord.incremental(cols, binlogTso,
binlogLsn, binlogOp)
+ ? DorisSourceRecord.incremental(cols, changelogTso,
changelogLsn, changelogOp)
: DorisSourceRecord.snapshot(cols);
}
}
@@ -119,9 +119,9 @@ public class RowBatch {
private RootAllocator rootAllocator;
private Schema schema;
private final boolean incremental;
- private static final String BINLOG_TSO = "__DORIS_BINLOG_TSO__";
- private static final String BINLOG_LSN = "__DORIS_BINLOG_LSN__";
- private static final String BINLOG_OP = "__DORIS_BINLOG_OP__";
+ private static final String CHANGELOG_TSO = "__DORIS_CHANGELOG_TSO__";
+ private static final String CHANGELOG_LSN = "__DORIS_CHANGELOG_LSN__";
+ private static final String CHANGELOG_OP = "__DORIS_CHANGELOG_OP__";
private static final String DATETIME_PATTERN = "yyyy-MM-dd HH:mm:ss";
private static final String DATETIMEV2_PATTERN = "yyyy-MM-dd
HH:mm:ss.SSSSSS";
private static final String DATE_PATTERN = "yyyy-MM-dd";
@@ -270,9 +270,9 @@ public class RowBatch {
for (int col = 0; col < fieldVectors.size(); col++) {
FieldVector fieldVector = fieldVectors.get(col);
String vectorName = fieldVector.getName();
- if (incremental && isBinlogField(vectorName)) {
+ if (incremental && isChangelogField(vectorName)) {
for (int rowIndex = 0; rowIndex < rowCountInOneBatch;
rowIndex++) {
- setBinlogValue(rowIndex, vectorName,
fieldVector.getObject(rowIndex));
+ setChangelogValue(rowIndex, vectorName,
fieldVector.getObject(rowIndex));
}
continue;
}
@@ -304,10 +304,10 @@ public class RowBatch {
}
}
- private static boolean isBinlogField(String name) {
- return BINLOG_TSO.equalsIgnoreCase(name)
- || BINLOG_LSN.equalsIgnoreCase(name)
- || BINLOG_OP.equalsIgnoreCase(name);
+ private static boolean isChangelogField(String name) {
+ return CHANGELOG_TSO.equalsIgnoreCase(name)
+ || CHANGELOG_LSN.equalsIgnoreCase(name)
+ || CHANGELOG_OP.equalsIgnoreCase(name);
}
private void validateIncrementalFields() throws DorisException {
@@ -316,9 +316,9 @@ public class RowBatch {
int opFields = 0;
for (FieldVector fieldVector : fieldVectors) {
String name = fieldVector.getName();
- tsoFields += BINLOG_TSO.equalsIgnoreCase(name) ? 1 : 0;
- lsnFields += BINLOG_LSN.equalsIgnoreCase(name) ? 1 : 0;
- opFields += BINLOG_OP.equalsIgnoreCase(name) ? 1 : 0;
+ tsoFields += CHANGELOG_TSO.equalsIgnoreCase(name) ? 1 : 0;
+ lsnFields += CHANGELOG_LSN.equalsIgnoreCase(name) ? 1 : 0;
+ opFields += CHANGELOG_OP.equalsIgnoreCase(name) ? 1 : 0;
}
if (tsoFields != 1 || lsnFields != 1 || opFields != 1) {
throw new DorisException(
@@ -326,9 +326,9 @@ public class RowBatch {
}
}
- private void setBinlogValue(int rowIndex, String name, Object value) {
+ private void setChangelogValue(int rowIndex, String name, Object value) {
if (value == null) {
- throw new DorisRuntimeException("Doris row-binlog field " + name +
" cannot be null");
+ throw new DorisRuntimeException("Doris changelog field " + name +
" cannot be null");
}
final long parsed;
try {
@@ -338,15 +338,15 @@ public class RowBatch {
: Long.parseLong(value.toString());
} catch (NumberFormatException e) {
throw new DorisRuntimeException(
- "Invalid Doris row-binlog field " + name + ": " + value,
e);
+ "Invalid Doris changelog field " + name + ": " + value, e);
}
Row row = rowBatch.get(readRowCount + rowIndex);
- if (BINLOG_TSO.equalsIgnoreCase(name)) {
- row.binlogTso = parsed;
- } else if (BINLOG_LSN.equalsIgnoreCase(name)) {
- row.binlogLsn = parsed;
- } else if (BINLOG_OP.equalsIgnoreCase(name)) {
- row.binlogOp = parsed;
+ if (CHANGELOG_TSO.equalsIgnoreCase(name)) {
+ row.changelogTso = parsed;
+ } else if (CHANGELOG_LSN.equalsIgnoreCase(name)) {
+ row.changelogLsn = parsed;
+ } else if (CHANGELOG_OP.equalsIgnoreCase(name)) {
+ row.changelogOp = parsed;
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisBinlogIncrementType.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisChangelogIncrementType.java
similarity index 76%
rename from
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisBinlogIncrementType.java
rename to
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisChangelogIncrementType.java
index 0a0dd809..c1efff5d 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisBinlogIncrementType.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/DorisChangelogIncrementType.java
@@ -19,21 +19,21 @@ package org.apache.doris.flink.source;
import java.util.Locale;
-/** Increment semantics requested from the Doris row-binlog query. */
-public enum DorisBinlogIncrementType {
+/** Increment semantics requested from the Doris changelog query. */
+public enum DorisChangelogIncrementType {
DETAIL,
MIN_DELTA,
APPEND_ONLY;
- public static DorisBinlogIncrementType fromOption(String value) {
+ public static DorisChangelogIncrementType fromOption(String value) {
if (value == null) {
- throw new IllegalArgumentException("source.binlog.increment-type
must not be null");
+ throw new
IllegalArgumentException("source.changelog.increment-type must not be null");
}
try {
return valueOf(value.trim().toUpperCase(Locale.ROOT));
} catch (java.lang.IllegalArgumentException e) {
throw new IllegalArgumentException(
- "Unsupported source.binlog.increment-type: " + value, e);
+ "Unsupported source.changelog.increment-type: " + value,
e);
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
index 5fbc594f..e2a3b271 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
@@ -73,20 +73,20 @@ public class DorisSourceSplitAssigner {
private static Optional<String> readExternalOffset(
DorisOptions options, DorisReadOptions readOptions) {
if (!readOptions.getScanMode().hasIncrementalPhase()
- || readOptions.getBinlogOffsetTable() == null) {
+ || readOptions.getChangelogOffsetTable() == null) {
return Optional.empty();
}
try (DorisOffsetPublisher publisher =
new DorisOffsetPublisher(
new SimpleJdbcConnectionProvider(options),
- readOptions.getBinlogOffsetTable(),
- readOptions.getBinlogConsumerId())) {
+ readOptions.getChangelogOffsetTable(),
+ readOptions.getChangelogConsumerId())) {
Optional<String> offset = publisher.readOffset();
LOG.info(
"Read external offset {} from {} for consumer {}.",
offset.orElse("<not found>"),
- readOptions.getBinlogOffsetTable(),
- readOptions.getBinlogConsumerId());
+ readOptions.getChangelogOffsetTable(),
+ readOptions.getChangelogConsumerId());
return offset;
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
index 2dd0f2dc..080aa998 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisFlightValueReader.java
@@ -43,7 +43,7 @@ import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.rest.SchemaUtils;
import org.apache.doris.flink.rest.models.Schema;
import org.apache.doris.flink.serialization.RowBatch;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.split.DorisSnapshotSplit;
import org.apache.doris.flink.source.split.DorisSourceSplit;
import org.apache.doris.flink.source.split.DorisStreamSplit;
@@ -119,14 +119,14 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
options,
readOptions,
(DorisStreamSplit) split,
- readOptions.getBinlogIncrementType()));
+ readOptions.getChangelogIncrementType()));
} else {
throw new DorisRuntimeException("Unknown Doris split type: " +
split);
}
this.queryResult =
split instanceof DorisStreamSplit
? executeQueryWithRetry(
- statement,
readOptions.getBinlogVisibleWaitTimeoutMs())
+ statement,
readOptions.getChangelogVisibleWaitTimeoutMs())
: statement.executeQuery();
this.arrowReader = queryResult.getReader();
} catch (InterruptedException e) {
@@ -215,7 +215,7 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
DorisOptions options,
DorisReadOptions readOptions,
DorisStreamSplit split,
- DorisBinlogIncrementType incrementType) {
+ DorisChangelogIncrementType incrementType) {
String[] tableIdentifiers = parseIdentifier(options);
String readFields =
StringUtils.isBlank(readOptions.getReadFields())
@@ -225,7 +225,7 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
PREFIX
+ " SELECT "
+ readFields
- + ", __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__,
__DORIS_BINLOG_OP__ FROM "
+ + ", __DORIS_CHANGELOG_TSO__, __DORIS_CHANGELOG_LSN__,
__DORIS_CHANGELOG_OP__ FROM "
+ quoteTable(tableIdentifiers)
+ "@incr('startTimestamp' = "
+ quoteLiteral(split.getStartTimestamp())
@@ -233,7 +233,7 @@ public class DorisFlightValueReader extends ValueReader
implements AutoCloseable
+ quoteLiteral(split.getEndTimestamp())
+ ", 'incrementType' = "
+ quoteLiteral(incrementType.toSqlValue())
- + ") ORDER BY __DORIS_BINLOG_TSO__,
__DORIS_BINLOG_LSN__, __DORIS_BINLOG_OP__";
+ + ") ORDER BY __DORIS_CHANGELOG_TSO__,
__DORIS_CHANGELOG_LSN__, __DORIS_CHANGELOG_OP__";
LOG.debug("Query SQL Sending to Doris FE is: '{}'.", sql);
return sql;
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceRecord.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceRecord.java
index eb89945a..a16ea0c6 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceRecord.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceRecord.java
@@ -26,22 +26,26 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
-/** A single Doris row and its row-binlog metadata. */
+/** A single Doris row and its Doris changelog metadata. */
@PublicEvolving
public final class DorisSourceRecord {
private final List<?> fieldValues;
private final RowKind rowKind;
- private final Long binlogTso;
- private final Long binlogLsn;
- private final Long binlogOp;
+ private final Long changelogTso;
+ private final Long changelogLsn;
+ private final Long changelogOp;
public DorisSourceRecord(
- List<?> fieldValues, RowKind rowKind, Long binlogTso, Long
binlogLsn, Long binlogOp) {
+ List<?> fieldValues,
+ RowKind rowKind,
+ Long changelogTso,
+ Long changelogLsn,
+ Long changelogOp) {
this.fieldValues = new ArrayList<>(fieldValues);
this.rowKind = Objects.requireNonNull(rowKind, "rowKind");
- this.binlogTso = binlogTso;
- this.binlogLsn = binlogLsn;
- this.binlogOp = binlogOp;
+ this.changelogTso = changelogTso;
+ this.changelogLsn = changelogLsn;
+ this.changelogOp = changelogOp;
}
public static DorisSourceRecord snapshot(List<?> fieldValues) {
@@ -50,7 +54,7 @@ public final class DorisSourceRecord {
public static DorisSourceRecord incremental(List<?> fieldValues, Long tso,
Long lsn, Long op) {
if (tso == null || lsn == null || op == null) {
- throw new DorisRuntimeException("Doris row-binlog metadata cannot
be null");
+ throw new DorisRuntimeException("Doris changelog metadata cannot
be null");
}
return new DorisSourceRecord(fieldValues, toRowKind(op), tso, lsn, op);
}
@@ -68,7 +72,7 @@ public final class DorisSourceRecord {
if (op == 3L) {
return RowKind.UPDATE_AFTER;
}
- throw new DorisRuntimeException("Unsupported Doris binlog operation: "
+ op);
+ throw new DorisRuntimeException("Unsupported Doris changelog
operation: " + op);
}
public List<?> getFieldValues() {
@@ -79,15 +83,15 @@ public final class DorisSourceRecord {
return rowKind;
}
- public Long getBinlogTso() {
- return binlogTso;
+ public Long getChangelogTso() {
+ return changelogTso;
}
- public Long getBinlogLsn() {
- return binlogLsn;
+ public Long getChangelogLsn() {
+ return changelogLsn;
}
- public Long getBinlogOp() {
- return binlogOp;
+ public Long getChangelogOp() {
+ return changelogOp;
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceSplitReader.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceSplitReader.java
index 5fef6344..2124da97 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceSplitReader.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisSourceSplitReader.java
@@ -53,12 +53,12 @@ public class DorisSourceSplitReader implements
SplitReader<DorisSourceRecord, Do
this.readOptions = readOptions;
this.splits = new ArrayDeque<>();
this.offsetPublisher =
- readOptions.getBinlogOffsetTable() == null
+ readOptions.getChangelogOffsetTable() == null
? null
: new DorisOffsetPublisher(
new SimpleJdbcConnectionProvider(options),
- readOptions.getBinlogOffsetTable(),
- readOptions.getBinlogConsumerId());
+ readOptions.getChangelogOffsetTable(),
+ readOptions.getChangelogConsumerId());
}
@Override
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/split/DorisStreamSplit.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/split/DorisStreamSplit.java
index f4e3ca66..2f8e796b 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/split/DorisStreamSplit.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/split/DorisStreamSplit.java
@@ -20,7 +20,7 @@ package org.apache.doris.flink.source.split;
import java.util.Objects;
import java.util.regex.Pattern;
-/** A finite Doris row-binlog query range with an inclusive start and
exclusive end. */
+/** A finite Doris changelog query range with an inclusive start and exclusive
end. */
public final class DorisStreamSplit implements DorisSourceSplit {
private static final Pattern TIMESTAMP_PATTERN =
Pattern.compile("\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}");
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
index 3cd66764..8300cd08 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
@@ -43,7 +43,7 @@ import static
org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_REQUEST_READ
import static
org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_REQUEST_RETRIES_DEFAULT;
import static
org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_TABLET_SIZE_DEFAULT;
import static
org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_THRIFT_MAX_MESSAGE_SIZE_DEFAULT;
-import static
org.apache.doris.flink.cfg.ConfigurationOptions.SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT;
+import static
org.apache.doris.flink.cfg.ConfigurationOptions.SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT;
import static org.apache.doris.flink.sink.writer.LoadConstants.COMPRESS_TYPE;
import static
org.apache.doris.flink.sink.writer.LoadConstants.COMPRESS_TYPE_GZ;
import static org.apache.doris.flink.sink.writer.LoadConstants.FORMAT_KEY;
@@ -421,36 +421,37 @@ public class DorisConfigOptions {
.noDefaultValue()
.withDescription(
"Inclusive first-start timestamp for
from-timestamp mode; a saved external offset takes precedence");
- public static final ConfigOption<String> SOURCE_BINLOG_INCREMENT_TYPE =
- ConfigOptions.key("source.binlog.increment-type")
+ public static final ConfigOption<String> SOURCE_CHANGELOG_INCREMENT_TYPE =
+ ConfigOptions.key("source.changelog.increment-type")
.stringType()
.defaultValue("detail")
- .withDescription("Doris row-binlog increment type");
- public static final ConfigOption<Duration> SOURCE_BINLOG_POLL_INTERVAL =
- ConfigOptions.key("source.binlog.poll-interval")
+ .withDescription("Doris changelog increment type");
+ public static final ConfigOption<Duration> SOURCE_CHANGELOG_POLL_INTERVAL =
+ ConfigOptions.key("source.changelog.poll-interval")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
"Interval between attempts to discover the next
Stream split; must be "
+ "at least 1 second");
- public static final ConfigOption<Duration>
SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT =
- ConfigOptions.key("source.binlog.visible-wait-timeout")
+ public static final ConfigOption<Duration>
SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT =
+ ConfigOptions.key("source.changelog.visible-wait-timeout")
.durationType()
-
.defaultValue(Duration.ofMillis(SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT))
+ .defaultValue(
+
Duration.ofMillis(SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT_MS_DEFAULT))
.withDescription(
"Maximum time to retry an incremental query after
Doris reports a "
+ "visible wait timeout; 0s disables
retries");
- public static final ConfigOption<String> SOURCE_BINLOG_OFFSET_TABLE =
- ConfigOptions.key("source.binlog.offset-table")
+ public static final ConfigOption<String> SOURCE_CHANGELOG_OFFSET_TABLE =
+ ConfigOptions.key("source.changelog.offset-table")
.stringType()
.noDefaultValue()
.withDescription(
"Doris table used to persist completed checkpoint
offsets and resume new jobs without restored Source state");
- public static final ConfigOption<String> SOURCE_BINLOG_CONSUMER_ID =
- ConfigOptions.key("source.binlog.consumer-id")
+ public static final ConfigOption<String> SOURCE_CHANGELOG_CONSUMER_ID =
+ ConfigOptions.key("source.changelog.consumer-id")
.stringType()
.noDefaultValue()
- .withDescription("Stable consumer identifier in the binlog
offset table");
+ .withDescription("Stable consumer identifier in the
changelog offset table");
public static final ConfigOption<Integer> FLIGHT_SQL_PORT =
ConfigOptions.key("source.flight-sql-port")
.intType()
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/DorisIncrementalRowBatchTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/DorisIncrementalRowBatchTest.java
index 671a6a63..75fdce8f 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/DorisIncrementalRowBatchTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/serialization/DorisIncrementalRowBatchTest.java
@@ -42,9 +42,9 @@ class DorisIncrementalRowBatchTest {
@Test
void rejectsIncrementalBatchWithMissingNamedHiddenField() throws Exception
{
try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE);
- BigIntVector lsn = new BigIntVector("__DORIS_BINLOG_LSN__",
allocator);
+ BigIntVector lsn = new BigIntVector("__DORIS_CHANGELOG_LSN__",
allocator);
IntVector id = new IntVector("id", allocator);
- BigIntVector op = new BigIntVector("__DORIS_BINLOG_OP__",
allocator);
+ BigIntVector op = new BigIntVector("__DORIS_CHANGELOG_OP__",
allocator);
VarCharVector name = new VarCharVector("name", allocator);
BigIntVector wrongTso = new BigIntVector("not_the_tso",
allocator);
VectorSchemaRoot root = VectorSchemaRoot.of(lsn, id, op, name,
wrongTso)) {
@@ -73,11 +73,11 @@ class DorisIncrementalRowBatchTest {
@Test
void extractsHiddenColumnsByNameInsteadOfPosition() throws Exception {
try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE);
- BigIntVector lsn = new BigIntVector("__DORIS_BINLOG_LSN__",
allocator);
+ BigIntVector lsn = new BigIntVector("__DORIS_CHANGELOG_LSN__",
allocator);
IntVector id = new IntVector("id", allocator);
- BigIntVector op = new BigIntVector("__DORIS_BINLOG_OP__",
allocator);
+ BigIntVector op = new BigIntVector("__DORIS_CHANGELOG_OP__",
allocator);
VarCharVector name = new VarCharVector("name", allocator);
- BigIntVector tso = new BigIntVector("__DORIS_BINLOG_TSO__",
allocator);
+ BigIntVector tso = new BigIntVector("__DORIS_CHANGELOG_TSO__",
allocator);
VectorSchemaRoot root = VectorSchemaRoot.of(lsn, id, op, name,
tso)) {
lsn.setSafe(0, 1001L);
id.setSafe(0, 7);
@@ -97,9 +97,9 @@ class DorisIncrementalRowBatchTest {
assertThat(record.getFieldValues()).isEqualTo(Arrays.asList(7,
"before"));
assertThat(record.getRowKind()).isEqualTo(RowKind.UPDATE_BEFORE);
- assertThat(record.getBinlogTso()).isEqualTo(101L);
- assertThat(record.getBinlogLsn()).isEqualTo(1001L);
- assertThat(record.getBinlogOp()).isEqualTo(2L);
+ assertThat(record.getChangelogTso()).isEqualTo(101L);
+ assertThat(record.getChangelogLsn()).isEqualTo(1001L);
+ assertThat(record.getChangelogOp()).isEqualTo(2L);
}
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
index dcf39914..3a1f8ba5 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
@@ -247,8 +247,8 @@ class DorisSourceOffsetRestoreTest {
return DorisReadOptions.builder()
.setScanMode(mode)
.setScanTimestamp(mode == DorisSourceScanMode.FROM_TIMESTAMP ?
CONFIGURED : null)
- .setBinlogOffsetTable("ops.offsets")
- .setBinlogConsumerId("orders-consumer")
+ .setChangelogOffsetTable("ops.offsets")
+ .setChangelogConsumerId("orders-consumer")
.build();
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisHybridSourceEnumeratorTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisHybridSourceEnumeratorTest.java
index af603866..878a6403 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisHybridSourceEnumeratorTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisHybridSourceEnumeratorTest.java
@@ -433,7 +433,7 @@ class DorisHybridSourceEnumeratorTest {
timestampSupplier,
snapshots);
return new DorisSourceEnumerator(
- context, splitAssigner,
testReadOptions.getBinlogPollIntervalMs());
+ context, splitAssigner,
testReadOptions.getChangelogPollIntervalMs());
}
private static DorisSourceEnumerator restoredEnumerator(
@@ -450,7 +450,7 @@ class DorisHybridSourceEnumeratorTest {
context.currentParallelism(),
timestampSupplier);
return new DorisSourceEnumerator(
- context, splitAssigner,
testReadOptions.getBinlogPollIntervalMs());
+ context, splitAssigner,
testReadOptions.getChangelogPollIntervalMs());
}
private static DorisOptions dorisOptions() {
@@ -463,7 +463,7 @@ class DorisHybridSourceEnumeratorTest {
private static DorisReadOptions readOptions(
DorisSourceScanMode mode, String configuredStartTimestamp) {
DorisReadOptions.Builder builder =
-
DorisReadOptions.builder().setScanMode(mode).setBinlogPollIntervalMs(10_000L);
+
DorisReadOptions.builder().setScanMode(mode).setChangelogPollIntervalMs(10_000L);
if (mode == DorisSourceScanMode.FROM_TIMESTAMP) {
builder.setScanTimestamp(configuredStartTimestamp);
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisSourceEnumeratorTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisSourceEnumeratorTest.java
index f0ca5bc5..bca659cb 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisSourceEnumeratorTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/enumerator/DorisSourceEnumeratorTest.java
@@ -126,7 +126,7 @@ public class DorisSourceEnumeratorTest {
.build(),
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.SNAPSHOT)
- .setBinlogPollIntervalMs(10_000L)
+ .setChangelogPollIntervalMs(10_000L)
.build(),
checkpoint,
context.currentParallelism());
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisFlightValueReaderTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisFlightValueReaderTest.java
index bdf39a2d..145c806a 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisFlightValueReaderTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisFlightValueReaderTest.java
@@ -35,7 +35,7 @@ import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.PartitionDefinition;
import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.rest.models.Schema;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.split.DorisSnapshotSplit;
import org.apache.doris.flink.source.split.DorisStreamSplit;
import org.junit.jupiter.api.Test;
@@ -180,7 +180,7 @@ class DorisFlightValueReaderTest {
String sql =
DorisFlightValueReader.buildIncrementalSql(
- options, readOptions, split,
DorisBinlogIncrementType.DETAIL);
+ options, readOptions, split,
DorisChangelogIncrementType.DETAIL);
assertThat(sql)
.contains("`sales`.`orders`@incr")
@@ -188,7 +188,7 @@ class DorisFlightValueReaderTest {
.contains("'endTimestamp' = '2026-07-20 10:00:10'")
.contains("'incrementType' = 'DETAIL'")
.endsWith(
- "ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__,
__DORIS_BINLOG_OP__");
+ "ORDER BY __DORIS_CHANGELOG_TSO__,
__DORIS_CHANGELOG_LSN__, __DORIS_CHANGELOG_OP__");
}
@Test
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisRecordEmitterTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisRecordEmitterTest.java
index 903f7bb3..76d7c356 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisRecordEmitterTest.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/reader/DorisRecordEmitterTest.java
@@ -78,7 +78,7 @@ class DorisRecordEmitterTest {
}
@Test
- void mapsAllBinlogOperationsAndRejectsUnknownValues() {
+ void mapsAllChangelogOperationsAndRejectsUnknownValues() {
assertThat(record(0L).getRowKind()).isEqualTo(RowKind.INSERT);
assertThat(record(1L).getRowKind()).isEqualTo(RowKind.DELETE);
assertThat(record(2L).getRowKind()).isEqualTo(RowKind.UPDATE_BEFORE);
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/resources/container/e2e/doris2doris/test_doris_incremental_all_types_source_tbl.sql
b/flink-doris-connector/flink-doris-connector-base/src/test/resources/container/e2e/doris2doris/test_doris_incremental_all_types_source_tbl.sql
index 61927f4f..a0e800f8 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/resources/container/e2e/doris2doris/test_doris_incremental_all_types_source_tbl.sql
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/resources/container/e2e/doris2doris/test_doris_incremental_all_types_source_tbl.sql
@@ -30,7 +30,6 @@ PROPERTIES (
"replication_num" = "1",
"light_schema_change" = "true",
"enable_unique_key_merge_on_write" = "true",
- "binlog.enable" = "true",
- "binlog.format" = "ROW",
- "binlog.need_historical_value" = "true"
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true"
);
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
index 49c3625f..1f7d6f89 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/DorisSource.java
@@ -49,13 +49,13 @@ import
org.apache.doris.flink.source.split.DorisSourceSplitSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** FLIP-27 source for bounded snapshots and continuous Doris row-binlog
increments. */
+/** FLIP-27 source for bounded snapshots and continuous Doris changelog
increments. */
@PublicEvolving
public class DorisSource<OUT>
implements Source<OUT, DorisSourceSplit, DorisSourceCheckpoint>,
ResultTypeQueryable<OUT> {
private static final Logger LOG =
LoggerFactory.getLogger(DorisSource.class);
- private static final long MIN_BINLOG_POLL_INTERVAL_MS = 1_000L;
+ private static final long MIN_CHANGELOG_POLL_INTERVAL_MS = 1_000L;
private final DorisOptions options;
private final DorisReadOptions readOptions;
@@ -102,7 +102,7 @@ public class DorisSource<OUT>
return new DorisSourceEnumerator(
context,
DorisSourceSplitAssigner.create(options, readOptions,
context.currentParallelism()),
- readOptions.getBinlogPollIntervalMs());
+ readOptions.getChangelogPollIntervalMs());
}
@Override
@@ -114,7 +114,7 @@ public class DorisSource<OUT>
context,
DorisSourceSplitAssigner.restore(
options, readOptions, checkpoint,
context.currentParallelism()),
- readOptions.getBinlogPollIntervalMs());
+ readOptions.getChangelogPollIntervalMs());
}
@Override
@@ -228,23 +228,23 @@ public class DorisSource<OUT>
"incremental source modes require
source.use-flight-sql=true");
}
Preconditions.checkArgument(
- readOptions.getBinlogPollIntervalMs() >=
MIN_BINLOG_POLL_INTERVAL_MS,
- "source.binlog.poll-interval must be at least 1 second");
+ readOptions.getChangelogPollIntervalMs() >=
MIN_CHANGELOG_POLL_INTERVAL_MS,
+ "source.changelog.poll-interval must be at least 1
second");
Preconditions.checkArgument(
- readOptions.getBinlogVisibleWaitTimeoutMs() >= 0,
- "source.binlog.visible-wait-timeout must not be negative");
- String offsetTable = readOptions.getBinlogOffsetTable();
- String consumerId = readOptions.getBinlogConsumerId();
+ readOptions.getChangelogVisibleWaitTimeoutMs() >= 0,
+ "source.changelog.visible-wait-timeout must not be
negative");
+ String offsetTable = readOptions.getChangelogOffsetTable();
+ String consumerId = readOptions.getChangelogConsumerId();
Preconditions.checkArgument(
(offsetTable == null) == (consumerId == null),
- "source.binlog.offset-table and source.binlog.consumer-id
must be configured together");
+ "source.changelog.offset-table and
source.changelog.consumer-id must be configured together");
if (offsetTable != null) {
Preconditions.checkArgument(
readOptions.getScanMode().hasIncrementalPhase(),
- "source.binlog offset persistence is only valid in
incremental source modes");
+ "source.changelog offset persistence is only valid in
incremental source modes");
Preconditions.checkNotNull(
options.getJdbcUrl(),
- "jdbc-url is required when source.binlog.offset-table
is configured");
+ "jdbc-url is required when
source.changelog.offset-table is configured");
try (DorisOffsetPublisher publisher =
new DorisOffsetPublisher(
new SimpleJdbcConnectionProvider(options),
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
index 8aab56f4..d0484b89 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
@@ -63,7 +63,7 @@ public class DorisSourceReader<T>
Configuration config) {
super(elementsQueue, fetcherManager, recordEmitter, config, context);
offsetPublishingEnabled =
- readOptions.getBinlogOffsetTable() != null &&
context.getIndexOfSubtask() == 0;
+ readOptions.getChangelogOffsetTable() != null &&
context.getIndexOfSubtask() == 0;
}
@Override
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
index ac053468..beaea7e3 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
@@ -33,7 +33,7 @@ import org.apache.doris.flink.cfg.DorisLookupOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.writer.WriteMode;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSourceScanMode;
import java.util.HashSet;
@@ -97,11 +97,11 @@ import static
org.apache.doris.flink.table.DorisConfigOptions.SINK_S3_ROLE_ARN;
import static
org.apache.doris.flink.table.DorisConfigOptions.SINK_S3_SECRET_KEY;
import static org.apache.doris.flink.table.DorisConfigOptions.SINK_USE_CACHE;
import static org.apache.doris.flink.table.DorisConfigOptions.SINK_WRITE_MODE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_CONSUMER_ID;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_INCREMENT_TYPE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_OFFSET_TABLE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_POLL_INTERVAL;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_CONSUMER_ID;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_INCREMENT_TYPE;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_OFFSET_TABLE;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_POLL_INTERVAL;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT;
import static org.apache.doris.flink.table.DorisConfigOptions.SOURCE_SCAN_MODE;
import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_SCAN_TIMESTAMP;
import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_USE_OLD_API;
@@ -189,11 +189,11 @@ public final class DorisDynamicTableFactory
options.add(SOURCE_USE_OLD_API);
options.add(SOURCE_SCAN_MODE);
options.add(SOURCE_SCAN_TIMESTAMP);
- options.add(SOURCE_BINLOG_INCREMENT_TYPE);
- options.add(SOURCE_BINLOG_POLL_INTERVAL);
- options.add(SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT);
- options.add(SOURCE_BINLOG_OFFSET_TABLE);
- options.add(SOURCE_BINLOG_CONSUMER_ID);
+ options.add(SOURCE_CHANGELOG_INCREMENT_TYPE);
+ options.add(SOURCE_CHANGELOG_POLL_INTERVAL);
+ options.add(SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT);
+ options.add(SOURCE_CHANGELOG_OFFSET_TABLE);
+ options.add(SOURCE_CHANGELOG_CONSUMER_ID);
options.add(SINK_WRITE_MODE);
options.add(SINK_IGNORE_COMMIT_ERROR);
@@ -276,16 +276,17 @@ public final class DorisDynamicTableFactory
.setUseOldApi(readableConfig.get(SOURCE_USE_OLD_API))
.setScanMode(DorisSourceScanMode.fromOption(readableConfig.get(SOURCE_SCAN_MODE)))
.setScanTimestamp(readableConfig.getOptional(SOURCE_SCAN_TIMESTAMP).orElse(null))
- .setBinlogIncrementType(
- DorisBinlogIncrementType.fromOption(
-
readableConfig.get(SOURCE_BINLOG_INCREMENT_TYPE)))
-
.setBinlogPollIntervalMs(readableConfig.get(SOURCE_BINLOG_POLL_INTERVAL).toMillis())
- .setBinlogVisibleWaitTimeoutMs(
-
readableConfig.get(SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT).toMillis())
- .setBinlogOffsetTable(
-
readableConfig.getOptional(SOURCE_BINLOG_OFFSET_TABLE).orElse(null))
- .setBinlogConsumerId(
-
readableConfig.getOptional(SOURCE_BINLOG_CONSUMER_ID).orElse(null))
+ .setChangelogIncrementType(
+ DorisChangelogIncrementType.fromOption(
+
readableConfig.get(SOURCE_CHANGELOG_INCREMENT_TYPE)))
+ .setChangelogPollIntervalMs(
+
readableConfig.get(SOURCE_CHANGELOG_POLL_INTERVAL).toMillis())
+ .setChangelogVisibleWaitTimeoutMs(
+
readableConfig.get(SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT).toMillis())
+ .setChangelogOffsetTable(
+
readableConfig.getOptional(SOURCE_CHANGELOG_OFFSET_TABLE).orElse(null))
+ .setChangelogConsumerId(
+
readableConfig.getOptional(SOURCE_CHANGELOG_CONSUMER_ID).orElse(null))
.setUseFlightSql(readableConfig.get(USE_FLIGHT_SQL))
.setFlightSqlPort(readableConfig.get(FLIGHT_SQL_PORT));
return builder.build();
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
index 9125e32d..5e60066f 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
@@ -95,8 +95,8 @@ public final class DorisDynamicTableSource
@Override
public ChangelogMode getChangelogMode() {
if (!readOptions.getScanMode().hasIncrementalPhase()
- || readOptions.getBinlogIncrementType()
- ==
org.apache.doris.flink.source.DorisBinlogIncrementType.APPEND_ONLY) {
+ || readOptions.getChangelogIncrementType()
+ ==
org.apache.doris.flink.source.DorisChangelogIncrementType.APPEND_ONLY) {
return ChangelogMode.insertOnly();
}
return ChangelogMode.newBuilder()
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
index df5b5701..2afe538e 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/source/DorisSourceOptionsTest.java
@@ -54,10 +54,11 @@ class DorisSourceOptionsTest {
@Test
void parsesIncrementTypes() {
-
assertThat(DorisBinlogIncrementType.fromOption("detail").toSqlValue()).isEqualTo("DETAIL");
-
assertThat(DorisBinlogIncrementType.fromOption("min_delta").toSqlValue())
+
assertThat(DorisChangelogIncrementType.fromOption("detail").toSqlValue())
+ .isEqualTo("DETAIL");
+
assertThat(DorisChangelogIncrementType.fromOption("min_delta").toSqlValue())
.isEqualTo("MIN_DELTA");
-
assertThat(DorisBinlogIncrementType.fromOption("append_only").toSqlValue())
+
assertThat(DorisChangelogIncrementType.fromOption("append_only").toSqlValue())
.isEqualTo("APPEND_ONLY");
}
@@ -69,9 +70,9 @@ class DorisSourceOptionsTest {
assertThatThrownBy(() ->
DorisSourceScanMode.fromOption("from_timestamp"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("source.scan.mode");
- assertThatThrownBy(() ->
DorisBinlogIncrementType.fromOption("invalid"))
+ assertThatThrownBy(() ->
DorisChangelogIncrementType.fromOption("invalid"))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("source.binlog.increment-type");
+ .hasMessageContaining("source.changelog.increment-type");
}
@Test
@@ -80,11 +81,12 @@ class DorisSourceOptionsTest {
assertThat(options.getScanMode()).isEqualTo(DorisSourceScanMode.SNAPSHOT);
assertThat(options.getScanTimestamp()).isNull();
-
assertThat(options.getBinlogIncrementType()).isEqualTo(DorisBinlogIncrementType.DETAIL);
- assertThat(options.getBinlogPollIntervalMs()).isEqualTo(10_000L);
- assertThat(options.getBinlogVisibleWaitTimeoutMs()).isEqualTo(5 * 60 *
1000L);
- assertThat(options.getBinlogOffsetTable()).isNull();
- assertThat(options.getBinlogConsumerId()).isNull();
+ assertThat(options.getChangelogIncrementType())
+ .isEqualTo(DorisChangelogIncrementType.DETAIL);
+ assertThat(options.getChangelogPollIntervalMs()).isEqualTo(10_000L);
+ assertThat(options.getChangelogVisibleWaitTimeoutMs()).isEqualTo(5 *
60 * 1000L);
+ assertThat(options.getChangelogOffsetTable()).isNull();
+ assertThat(options.getChangelogConsumerId()).isNull();
}
@Test
@@ -93,11 +95,11 @@ class DorisSourceOptionsTest {
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
-
.setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
- .setBinlogVisibleWaitTimeoutMs(60_000L)
- .setBinlogOffsetTable("ops.flink_source_offsets")
- .setBinlogConsumerId("prod.sales.orders")
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
+ .setChangelogVisibleWaitTimeoutMs(60_000L)
+ .setChangelogOffsetTable("ops.flink_source_offsets")
+ .setChangelogConsumerId("prod.sales.orders")
.build();
DorisReadOptions copy = options.copy();
@@ -105,11 +107,12 @@ class DorisSourceOptionsTest {
assertThat(copy).isEqualTo(options).isNotSameAs(options);
assertThat(copy.getScanMode()).isEqualTo(DorisSourceScanMode.FROM_TIMESTAMP);
assertThat(copy.getScanTimestamp()).isEqualTo("2026-07-20 10:00:00");
-
assertThat(copy.getBinlogIncrementType()).isEqualTo(DorisBinlogIncrementType.MIN_DELTA);
- assertThat(copy.getBinlogPollIntervalMs()).isEqualTo(3_000L);
- assertThat(copy.getBinlogVisibleWaitTimeoutMs()).isEqualTo(60_000L);
-
assertThat(copy.getBinlogOffsetTable()).isEqualTo("ops.flink_source_offsets");
- assertThat(copy.getBinlogConsumerId()).isEqualTo("prod.sales.orders");
+ assertThat(copy.getChangelogIncrementType())
+ .isEqualTo(DorisChangelogIncrementType.MIN_DELTA);
+ assertThat(copy.getChangelogPollIntervalMs()).isEqualTo(3_000L);
+ assertThat(copy.getChangelogVisibleWaitTimeoutMs()).isEqualTo(60_000L);
+
assertThat(copy.getChangelogOffsetTable()).isEqualTo("ops.flink_source_offsets");
+
assertThat(copy.getChangelogConsumerId()).isEqualTo("prod.sales.orders");
}
@Test
@@ -158,16 +161,20 @@ class DorisSourceOptionsTest {
() ->
buildSource(
DorisReadOptions.builder()
- .setBinlogPollIntervalMs(999L)
+
.setChangelogPollIntervalMs(999L)
.build()))
.hasMessageContaining("at least 1 second");
-
assertThat(buildSource(DorisReadOptions.builder().setBinlogPollIntervalMs(1_000L).build()))
+ assertThat(
+ buildSource(
+ DorisReadOptions.builder()
+ .setChangelogPollIntervalMs(1_000L)
+ .build()))
.isNotNull();
assertThatThrownBy(
() ->
buildSource(
DorisReadOptions.builder()
-
.setBinlogVisibleWaitTimeoutMs(-1L)
+
.setChangelogVisibleWaitTimeoutMs(-1L)
.build()))
.hasMessageContaining("visible-wait-timeout must not be
negative");
assertThat(
@@ -186,7 +193,7 @@ class DorisSourceOptionsTest {
buildSource(
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.LATEST)
-
.setBinlogOffsetTable("ops.flink_source_offsets")
+
.setChangelogOffsetTable("ops.flink_source_offsets")
.build()))
.hasMessageContaining("must be configured together");
assertThatThrownBy(
@@ -194,8 +201,8 @@ class DorisSourceOptionsTest {
buildSource(
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.SNAPSHOT)
-
.setBinlogOffsetTable("ops.flink_source_offsets")
-
.setBinlogConsumerId("prod.sales.orders")
+
.setChangelogOffsetTable("ops.flink_source_offsets")
+
.setChangelogConsumerId("prod.sales.orders")
.build()))
.hasMessageContaining("only valid in incremental source
modes");
assertThatThrownBy(
@@ -203,8 +210,8 @@ class DorisSourceOptionsTest {
buildSource(
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.LATEST)
-
.setBinlogOffsetTable("ops.flink_source_offsets")
-
.setBinlogConsumerId("prod.sales.orders")
+
.setChangelogOffsetTable("ops.flink_source_offsets")
+
.setChangelogConsumerId("prod.sales.orders")
.build()))
.hasMessageContaining("jdbc-url");
}
@@ -222,9 +229,9 @@ class DorisSourceOptionsTest {
buildSource(
DorisReadOptions.builder()
.setScanMode(DorisSourceScanMode.LATEST)
- .setBinlogOffsetTable(
+ .setChangelogOffsetTable(
"ops.flink_source_offsets")
-
.setBinlogConsumerId("prod.sales.orders")
+
.setChangelogConsumerId("prod.sales.orders")
.build(),
"db.table",
"jdbc:mysql://127.0.0.1:9030"))
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
index 298c6805..8f120a31 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
@@ -30,7 +30,7 @@ import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.cfg.S3TvfOptions;
import org.apache.doris.flink.sink.writer.WriteMode;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSourceScanMode;
import org.junit.Test;
@@ -98,9 +98,9 @@ public class DorisDynamicTableFactoryTest {
properties.put("source.flight-sql-port", "-1");
properties.put("source.scan.mode", "from-timestamp");
properties.put("source.scan.timestamp", "2026-07-20 10:00:00");
- properties.put("source.binlog.increment-type", "min_delta");
- properties.put("source.binlog.poll-interval", "3s");
- properties.put("source.binlog.visible-wait-timeout", "7m");
+ properties.put("source.changelog.increment-type", "min_delta");
+ properties.put("source.changelog.poll-interval", "3s");
+ properties.put("source.changelog.visible-wait-timeout", "7m");
DynamicTableSource actual = FactoryMocks.createTableSource(SCHEMA,
properties);
DorisOptions options =
DorisOptions.builder()
@@ -140,9 +140,9 @@ public class DorisDynamicTableFactoryTest {
.setThriftMaxMessageSize(DORIS_THRIFT_MAX_MESSAGE_SIZE_DEFAULT)
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
- .setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
- .setBinlogVisibleWaitTimeoutMs(7 * 60_000L);
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
+ .setChangelogVisibleWaitTimeoutMs(7 * 60_000L);
DorisDynamicTableSource expected =
new DorisDynamicTableSource(
options,
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
index 0e9e82f9..89384383 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
@@ -43,7 +43,7 @@ import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.rest.PartitionDefinition;
import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.sink.OptionUtils;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSource;
import org.apache.doris.flink.source.DorisSourceScanMode;
import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
@@ -158,7 +158,7 @@ public class DorisDynamicTableSourceTest {
DorisReadOptions appendOnly =
OptionUtils.dorisReadOptionsBuilder()
.setScanMode(DorisSourceScanMode.LATEST)
-
.setBinlogIncrementType(DorisBinlogIncrementType.APPEND_ONLY)
+
.setChangelogIncrementType(DorisChangelogIncrementType.APPEND_ONLY)
.build();
DorisDynamicTableSource appendOnlySource =
new DorisDynamicTableSource(
@@ -229,8 +229,8 @@ public class DorisDynamicTableSourceTest {
OptionUtils.dorisReadOptionsBuilder()
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
-
.setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
.build();
DorisDynamicTableSource tableSource =
new DorisDynamicTableSource(
@@ -252,8 +252,8 @@ public class DorisDynamicTableSourceTest {
.setReadFields("`a`, `c`")
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
-
.setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
.build();
DorisDynamicTableSource expected =
new DorisDynamicTableSource(
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
index 5c214ea1..5eb7ce95 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/DorisSource.java
@@ -46,13 +46,13 @@ import
org.apache.doris.flink.source.split.DorisSourceSplitSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** FLIP-27 source for bounded snapshots and continuous Doris row-binlog
increments. */
+/** FLIP-27 source for bounded snapshots and continuous Doris changelog
increments. */
@PublicEvolving
public class DorisSource<OUT>
implements Source<OUT, DorisSourceSplit, DorisSourceCheckpoint>,
ResultTypeQueryable<OUT> {
private static final Logger LOG =
LoggerFactory.getLogger(DorisSource.class);
- private static final long MIN_BINLOG_POLL_INTERVAL_MS = 1_000L;
+ private static final long MIN_CHANGELOG_POLL_INTERVAL_MS = 1_000L;
private final DorisOptions options;
private final DorisReadOptions readOptions;
@@ -97,7 +97,7 @@ public class DorisSource<OUT>
return new DorisSourceEnumerator(
context,
DorisSourceSplitAssigner.create(options, readOptions,
context.currentParallelism()),
- readOptions.getBinlogPollIntervalMs());
+ readOptions.getChangelogPollIntervalMs());
}
@Override
@@ -109,7 +109,7 @@ public class DorisSource<OUT>
context,
DorisSourceSplitAssigner.restore(
options, readOptions, checkpoint,
context.currentParallelism()),
- readOptions.getBinlogPollIntervalMs());
+ readOptions.getChangelogPollIntervalMs());
}
@Override
@@ -223,23 +223,23 @@ public class DorisSource<OUT>
"incremental source modes require
source.use-flight-sql=true");
}
Preconditions.checkArgument(
- readOptions.getBinlogPollIntervalMs() >=
MIN_BINLOG_POLL_INTERVAL_MS,
- "source.binlog.poll-interval must be at least 1 second");
+ readOptions.getChangelogPollIntervalMs() >=
MIN_CHANGELOG_POLL_INTERVAL_MS,
+ "source.changelog.poll-interval must be at least 1
second");
Preconditions.checkArgument(
- readOptions.getBinlogVisibleWaitTimeoutMs() >= 0,
- "source.binlog.visible-wait-timeout must not be negative");
- String offsetTable = readOptions.getBinlogOffsetTable();
- String consumerId = readOptions.getBinlogConsumerId();
+ readOptions.getChangelogVisibleWaitTimeoutMs() >= 0,
+ "source.changelog.visible-wait-timeout must not be
negative");
+ String offsetTable = readOptions.getChangelogOffsetTable();
+ String consumerId = readOptions.getChangelogConsumerId();
Preconditions.checkArgument(
(offsetTable == null) == (consumerId == null),
- "source.binlog.offset-table and source.binlog.consumer-id
must be configured together");
+ "source.changelog.offset-table and
source.changelog.consumer-id must be configured together");
if (offsetTable != null) {
Preconditions.checkArgument(
readOptions.getScanMode().hasIncrementalPhase(),
- "source.binlog offset persistence is only valid in
incremental source modes");
+ "source.changelog offset persistence is only valid in
incremental source modes");
Preconditions.checkNotNull(
options.getJdbcUrl(),
- "jdbc-url is required when source.binlog.offset-table
is configured");
+ "jdbc-url is required when
source.changelog.offset-table is configured");
try (DorisOffsetPublisher publisher =
new DorisOffsetPublisher(
new SimpleJdbcConnectionProvider(options),
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
index 7330fd54..1c0f7c5d 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/source/reader/DorisSourceReader.java
@@ -60,7 +60,7 @@ public class DorisSourceReader<T>
Configuration config) {
super(fetcherManager, recordEmitter, config, context);
offsetPublishingEnabled =
- readOptions.getBinlogOffsetTable() != null &&
context.getIndexOfSubtask() == 0;
+ readOptions.getChangelogOffsetTable() != null &&
context.getIndexOfSubtask() == 0;
}
@Override
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
index e730923a..be335e51 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
@@ -33,7 +33,7 @@ import org.apache.doris.flink.cfg.DorisLookupOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.writer.WriteMode;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSourceScanMode;
import java.util.HashSet;
@@ -97,11 +97,11 @@ import static
org.apache.doris.flink.table.DorisConfigOptions.SINK_S3_ROLE_ARN;
import static
org.apache.doris.flink.table.DorisConfigOptions.SINK_S3_SECRET_KEY;
import static org.apache.doris.flink.table.DorisConfigOptions.SINK_USE_CACHE;
import static org.apache.doris.flink.table.DorisConfigOptions.SINK_WRITE_MODE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_CONSUMER_ID;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_INCREMENT_TYPE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_OFFSET_TABLE;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_POLL_INTERVAL;
-import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_CONSUMER_ID;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_INCREMENT_TYPE;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_OFFSET_TABLE;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_POLL_INTERVAL;
+import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT;
import static org.apache.doris.flink.table.DorisConfigOptions.SOURCE_SCAN_MODE;
import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_SCAN_TIMESTAMP;
import static
org.apache.doris.flink.table.DorisConfigOptions.SOURCE_USE_OLD_API;
@@ -189,11 +189,11 @@ public final class DorisDynamicTableFactory
options.add(SOURCE_USE_OLD_API);
options.add(SOURCE_SCAN_MODE);
options.add(SOURCE_SCAN_TIMESTAMP);
- options.add(SOURCE_BINLOG_INCREMENT_TYPE);
- options.add(SOURCE_BINLOG_POLL_INTERVAL);
- options.add(SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT);
- options.add(SOURCE_BINLOG_OFFSET_TABLE);
- options.add(SOURCE_BINLOG_CONSUMER_ID);
+ options.add(SOURCE_CHANGELOG_INCREMENT_TYPE);
+ options.add(SOURCE_CHANGELOG_POLL_INTERVAL);
+ options.add(SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT);
+ options.add(SOURCE_CHANGELOG_OFFSET_TABLE);
+ options.add(SOURCE_CHANGELOG_CONSUMER_ID);
options.add(SINK_WRITE_MODE);
options.add(SINK_IGNORE_COMMIT_ERROR);
@@ -276,16 +276,17 @@ public final class DorisDynamicTableFactory
.setUseOldApi(readableConfig.get(SOURCE_USE_OLD_API))
.setScanMode(DorisSourceScanMode.fromOption(readableConfig.get(SOURCE_SCAN_MODE)))
.setScanTimestamp(readableConfig.getOptional(SOURCE_SCAN_TIMESTAMP).orElse(null))
- .setBinlogIncrementType(
- DorisBinlogIncrementType.fromOption(
-
readableConfig.get(SOURCE_BINLOG_INCREMENT_TYPE)))
-
.setBinlogPollIntervalMs(readableConfig.get(SOURCE_BINLOG_POLL_INTERVAL).toMillis())
- .setBinlogVisibleWaitTimeoutMs(
-
readableConfig.get(SOURCE_BINLOG_VISIBLE_WAIT_TIMEOUT).toMillis())
- .setBinlogOffsetTable(
-
readableConfig.getOptional(SOURCE_BINLOG_OFFSET_TABLE).orElse(null))
- .setBinlogConsumerId(
-
readableConfig.getOptional(SOURCE_BINLOG_CONSUMER_ID).orElse(null))
+ .setChangelogIncrementType(
+ DorisChangelogIncrementType.fromOption(
+
readableConfig.get(SOURCE_CHANGELOG_INCREMENT_TYPE)))
+ .setChangelogPollIntervalMs(
+
readableConfig.get(SOURCE_CHANGELOG_POLL_INTERVAL).toMillis())
+ .setChangelogVisibleWaitTimeoutMs(
+
readableConfig.get(SOURCE_CHANGELOG_VISIBLE_WAIT_TIMEOUT).toMillis())
+ .setChangelogOffsetTable(
+
readableConfig.getOptional(SOURCE_CHANGELOG_OFFSET_TABLE).orElse(null))
+ .setChangelogConsumerId(
+
readableConfig.getOptional(SOURCE_CHANGELOG_CONSUMER_ID).orElse(null))
.setUseFlightSql(readableConfig.get(USE_FLIGHT_SQL))
.setFlightSqlPort(readableConfig.get(FLIGHT_SQL_PORT));
return builder.build();
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
index d5cf9506..a54c3163 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSource.java
@@ -95,8 +95,8 @@ public final class DorisDynamicTableSource
@Override
public ChangelogMode getChangelogMode() {
if (!readOptions.getScanMode().hasIncrementalPhase()
- || readOptions.getBinlogIncrementType()
- ==
org.apache.doris.flink.source.DorisBinlogIncrementType.APPEND_ONLY) {
+ || readOptions.getChangelogIncrementType()
+ ==
org.apache.doris.flink.source.DorisChangelogIncrementType.APPEND_ONLY) {
return ChangelogMode.insertOnly();
}
return ChangelogMode.newBuilder()
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
index 38a2b8fc..7f4e5633 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableFactoryTest.java
@@ -30,7 +30,7 @@ import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.cfg.S3TvfOptions;
import org.apache.doris.flink.sink.writer.WriteMode;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSourceScanMode;
import org.junit.Test;
@@ -98,9 +98,9 @@ public class DorisDynamicTableFactoryTest {
properties.put("source.flight-sql-port", "-1");
properties.put("source.scan.mode", "from-timestamp");
properties.put("source.scan.timestamp", "2026-07-20 10:00:00");
- properties.put("source.binlog.increment-type", "min_delta");
- properties.put("source.binlog.poll-interval", "3s");
- properties.put("source.binlog.visible-wait-timeout", "7m");
+ properties.put("source.changelog.increment-type", "min_delta");
+ properties.put("source.changelog.poll-interval", "3s");
+ properties.put("source.changelog.visible-wait-timeout", "7m");
DynamicTableSource actual = FactoryMocks.createTableSource(SCHEMA,
properties);
DorisOptions options =
DorisOptions.builder()
@@ -140,9 +140,9 @@ public class DorisDynamicTableFactoryTest {
.setThriftMaxMessageSize(DORIS_THRIFT_MAX_MESSAGE_SIZE_DEFAULT)
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
- .setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
- .setBinlogVisibleWaitTimeoutMs(7 * 60_000L);
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
+ .setChangelogVisibleWaitTimeoutMs(7 * 60_000L);
DorisDynamicTableSource expected =
new DorisDynamicTableSource(
options,
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
index 8464ee83..98c844f7 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/table/DorisDynamicTableSourceTest.java
@@ -43,7 +43,7 @@ import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.rest.PartitionDefinition;
import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.sink.OptionUtils;
-import org.apache.doris.flink.source.DorisBinlogIncrementType;
+import org.apache.doris.flink.source.DorisChangelogIncrementType;
import org.apache.doris.flink.source.DorisSource;
import org.apache.doris.flink.source.DorisSourceScanMode;
import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
@@ -158,7 +158,7 @@ public class DorisDynamicTableSourceTest {
DorisReadOptions appendOnly =
OptionUtils.dorisReadOptionsBuilder()
.setScanMode(DorisSourceScanMode.LATEST)
-
.setBinlogIncrementType(DorisBinlogIncrementType.APPEND_ONLY)
+
.setChangelogIncrementType(DorisChangelogIncrementType.APPEND_ONLY)
.build();
DorisDynamicTableSource appendOnlySource =
new DorisDynamicTableSource(
@@ -229,8 +229,8 @@ public class DorisDynamicTableSourceTest {
OptionUtils.dorisReadOptionsBuilder()
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
-
.setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
.build();
DorisDynamicTableSource tableSource =
new DorisDynamicTableSource(
@@ -252,8 +252,8 @@ public class DorisDynamicTableSourceTest {
.setReadFields("`a`, `c`")
.setScanMode(DorisSourceScanMode.FROM_TIMESTAMP)
.setScanTimestamp("2026-07-20 10:00:00")
-
.setBinlogIncrementType(DorisBinlogIncrementType.MIN_DELTA)
- .setBinlogPollIntervalMs(3_000L)
+
.setChangelogIncrementType(DorisChangelogIncrementType.MIN_DELTA)
+ .setChangelogPollIntervalMs(3_000L)
.build();
DorisDynamicTableSource expected =
new DorisDynamicTableSource(
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/container/e2e/DorisIncrementalSourceE2ECase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/container/e2e/DorisIncrementalSourceE2ECase.java
index 4d372081..10e9721b 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/container/e2e/DorisIncrementalSourceE2ECase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/container/e2e/DorisIncrementalSourceE2ECase.java
@@ -151,8 +151,8 @@ public class DorisIncrementalSourceE2ECase extends
AbstractITCaseService {
+ " 'source.scan.timestamp' = '%s',\n"
+ " 'source.use-flight-sql' = 'true',\n"
+ " 'source.flight-sql-port' = '%d',\n"
- + " 'source.binlog.increment-type' =
'detail',\n"
- + " 'source.binlog.poll-interval' =
'1s'\n"
+ + " 'source.changelog.increment-type' =
'detail',\n"
+ + " 'source.changelog.poll-interval' =
'1s'\n"
+ ")",
DorisConfigOptions.IDENTIFIER,
escapeSqlLiteral(getFenodes()),
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
index 4f7c470c..c795b19f 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
@@ -106,9 +106,8 @@ public abstract class AbstractDorisIncrementalITCase
extends AbstractITCaseServi
+ "PROPERTIES (\n"
+ " \"replication_num\" = \"1\",\n"
+ " \"enable_unique_key_merge_on_write\" =
\"true\",\n"
- + " \"binlog.enable\" = \"true\",\n"
- + " \"binlog.format\" = \"ROW\",\n"
- + " \"binlog.need_historical_value\" =
\"true\"\n"
+ + " \"changelog.enable\" = \"true\",\n"
+ + " \"changelog.need_historical_value\" =
\"true\"\n"
+ ")",
DATABASE, table));
if (values.length > 0) {
@@ -274,8 +273,8 @@ public abstract class AbstractDorisIncrementalITCase
extends AbstractITCaseServi
: String.format(
",\n"
+ " 'jdbc-url' = '%s',\n"
- + " 'source.binlog.offset-table' =
'%s.%s',\n"
- + " 'source.binlog.consumer-id' =
'%s'",
+ + " 'source.changelog.offset-table' =
'%s.%s',\n"
+ + " 'source.changelog.consumer-id' =
'%s'",
escapeSqlLiteral(getDorisQueryUrl()),
DATABASE,
offsetTable,
@@ -296,8 +295,8 @@ public abstract class AbstractDorisIncrementalITCase
extends AbstractITCaseServi
+ " 'source.scan.mode' = '%s',\n"
+ " 'source.use-flight-sql' = 'true',\n"
+ " 'source.flight-sql-port' = '%d',\n"
- + " 'source.binlog.increment-type' = 'detail',\n"
- + " 'source.binlog.poll-interval' = '1s'%s\n"
+ + " 'source.changelog.increment-type' =
'detail',\n"
+ + " 'source.changelog.poll-interval' = '1s'%s\n"
+ ")",
DorisConfigOptions.IDENTIFIER,
escapeSqlLiteral(getFenodes()),
@@ -326,8 +325,8 @@ public abstract class AbstractDorisIncrementalITCase
extends AbstractITCaseServi
+ " 'source.scan.timestamp' = '%s',\n"
+ " 'source.use-flight-sql' = 'true',\n"
+ " 'source.flight-sql-port' = '%d',\n"
- + " 'source.binlog.increment-type' = 'detail',\n"
- + " 'source.binlog.poll-interval' = '1s'%s\n"
+ + " 'source.changelog.increment-type' = 'detail',\n"
+ + " 'source.changelog.poll-interval' = '1s'%s\n"
+ ")",
DorisConfigOptions.IDENTIFIER,
escapeSqlLiteral(getFenodes()),
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
index 565d54c6..823b9922 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
@@ -87,7 +87,7 @@ public class DorisIncrementalSourceITCase extends
AbstractDorisIncrementalITCase
List<ObservedRow> observedRows = collector.getRows();
for (ObservedRow row : observedRows) {
Assert.assertEquals(
- "Hidden binlog fields escaped the projection", 2,
row.getArity());
+ "Hidden Doris changelog fields escaped the
projection", 2, row.getArity());
}
Assert.assertTrue(
"UPDATE_BEFORE must precede UPDATE_AFTER",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
