This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7e63fc99245 [fix](streaming-job) Avoid redundant schema persistence in
CDC offset journals (#69129)
7e63fc99245 is described below
commit 7e63fc99245545c908a8b81ab79f5785a82eedf0
Author: wudi <[email protected]>
AuthorDate: Fri Oct 16 11:14:54 2026 +0800
[fix](streaming-job) Avoid redundant schema persistence in CDC offset
journals (#69129)
### What problem does this PR solve?
Problem Summary:
FROM-TO CDC offset commits repeatedly serialize the complete cached
table schema into scheduler-job journals, even when the CDC client
reports no schema update. A roughly 1 MiB schema can therefore add
roughly 1 MiB to every ordinary offset journal.
Keep the complete provider state in the Job and checkpoint images, but
include schemas in offset journals only when the client supplies them.
Clear the pending-schema flag after journal acknowledgement so failed
writes can retry. Merge schema-omitting records during replay, including
cold ALTER and cloud offset recovery, while retaining full-replacement
semantics for TVF providers. Reuse the existing scheduler-job operation
and freeze the journal payload before queueing it.
Compatibility: the operation and outer encoding are unchanged, but older
FE binaries replace provider state wholesale. Replaying compact journals
with an old FE during a rolling upgrade or downgrade can clear its
cached schema. This change does not add a writer compatibility switch.
---
.../insert/streaming/StreamingInsertJob.java | 31 +-
.../doris/job/offset/SourceOffsetProvider.java | 5 +
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 46 ++-
.../offset/jdbc/JdbcTvfSourceOffsetProvider.java | 6 +
.../java/org/apache/doris/persist/EditLog.java | 4 +
.../StreamingInsertJobOffsetPersistenceTest.java | 5 +
.../StreamingInsertJobSchemaPersistenceTest.java | 434 +++++++++++++++++++++
.../jdbc/JdbcSourceOffsetProviderOffsetTest.java | 79 ++++
8 files changed, 601 insertions(+), 9 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index bf93b77af1e..028fa871767 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -85,6 +85,7 @@ import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TxnStateChangeCallback;
import com.google.common.base.Preconditions;
+import com.google.gson.JsonObject;
import com.google.gson.annotations.SerializedName;
import lombok.Getter;
import lombok.Setter;
@@ -562,7 +563,9 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
}
offsetProvider.updateOffset(offset);
offsetProvider.resetLag();
- this.offsetProviderPersist =
offsetProvider.getPersistInfo();
+ String persistInfo = offsetProvider.getPersistInfo();
+ // A paused provider may not have restored its schema from
the image yet.
+ this.offsetProviderPersist =
offsetProvider.mergePersistInfo(offsetProviderPersist, persistInfo);
log.info("Alter streaming job {} offset to {}",
getJobId(), offsetProperty);
}
modifyPropertiesInternal(alterJobCommand.getProperties());
@@ -1051,7 +1054,9 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
this.jobStatistic.setFileSize(attachment.getFileBytes());
this.jobStatistic.setFilteredRows(attachment.getFilteredRows());
offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset()));
- this.offsetProviderPersist = offsetProvider.getPersistInfo();
+ String persistInfo = offsetProvider.getPersistInfo();
+ // Preserve cached schema until the provider is restored.
+ this.offsetProviderPersist =
offsetProvider.mergePersistInfo(offsetProviderPersist, persistInfo);
//update metric
if (MetricRepo.isInit && !isReplay) {
@@ -1126,6 +1131,10 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
* @param replayJob
*/
public void replayOnUpdated(StreamingInsertJob replayJob) {
+ String replayPersistInfo = replayJob.getOffsetProviderPersist();
+ if (replayPersistInfo != null) {
+ replayPersistInfo =
offsetProvider.mergePersistInfo(offsetProviderPersist, replayPersistInfo);
+ }
if (!JobStatus.RUNNING.equals(replayJob.getJobStatus())) {
// No need to restore in the running state, as scheduling relies
on pending states.
// insert TVF does not persist the running state.
@@ -1138,8 +1147,8 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
}
try {
modifyPropertiesInternal(replayJob.getProperties());
- if (replayJob.getOffsetProviderPersist() != null) {
- setOffsetProviderPersist(replayJob.getOffsetProviderPersist());
+ if (replayPersistInfo != null) {
+ setOffsetProviderPersist(replayPersistInfo);
if (tvfType != null && !Config.isCloudMode()) {
offsetProvider.restoreFromPersistInfo(offsetProviderPersist);
}
@@ -1148,7 +1157,8 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
String offset =
replayJob.getProperties().get(StreamingJobProperties.OFFSET_PROPERTY);
if (StringUtils.isNotEmpty(offset)) {
offsetProvider.updateOffset(validateOffset(offset));
- this.offsetProviderPersist =
offsetProvider.getPersistInfo();
+ this.offsetProviderPersist =
offsetProvider.mergePersistInfo(
+ offsetProviderPersist,
offsetProvider.getPersistInfo());
}
}
// When the pause state is restarted, it also needs to be updated
@@ -1427,6 +1437,13 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
+ protected void logUpdateOperation(String persistInfo) {
+ JsonObject journalJson = GsonUtils.GSON.toJsonTree(this,
StreamingInsertJob.class).getAsJsonObject();
+ journalJson.addProperty("opp", persistInfo);
+ // Freeze only the journal payload; the job retains its complete state
for images.
+
Env.getCurrentEnv().getEditLog().logUpdateJob(GsonUtils.GSON.toJson(journalJson));
+ }
+
@Override
public long getId() {
return getJobId();
@@ -1761,7 +1778,9 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
return;
}
- logUpdateOperation();
+ logUpdateOperation(jdbcOffsetProvider.isTableSchemasDirty()
+ ? offsetProviderPersist :
jdbcOffsetProvider.getPersistInfoWithoutTableSchemas());
+ jdbcOffsetProvider.markTableSchemasPersisted();
lastOffsetPersistTimeMs = currentTimeMs;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
index 624e6a431dd..2962a672ec7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
@@ -150,6 +150,11 @@ public interface SourceOffsetProvider {
return null;
}
+ /** Merge serialized state without initializing the provider; full
replacement by default. */
+ default String mergePersistInfo(String previousPersistInfo, String
incomingPersistInfo) {
+ return incomingPersistInfo;
+ }
+
/**
* Restore offset from an image or job update journal, including for
PAUSED jobs.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
index 94cfd4fbbac..39c8ee71268 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
@@ -53,6 +53,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
+import com.google.gson.JsonObject;
import com.google.gson.annotations.SerializedName;
import lombok.Getter;
import lombok.Setter;
@@ -82,6 +83,7 @@ import java.util.stream.Collectors;
@Log4j2
public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
public static final String SPLIT_ID = "splitId";
+ private static final String TABLE_SCHEMAS_FIELD = "ts";
private static final ObjectMapper objectMapper = new ObjectMapper();
protected int snapshotParallelism = 1;
protected Long jobId;
@@ -100,8 +102,9 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
@SerializedName("bop")
Map<String, String> binlogOffsetPersist;
- @SerializedName("ts")
+ @SerializedName(TABLE_SCHEMAS_FIELD)
String tableSchemas;
+ private transient boolean tableSchemasDirty;
/** Split progress (task-commit view). */
@SerializedName("csp")
@@ -294,7 +297,7 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
public boolean shouldPersistOffset(long lastPersistTimeMs, long
currentTimeMs) {
synchronized (splitsLock) {
- if (currentOffset == null || !currentOffset.snapshotSplit()) {
+ if (tableSchemasDirty || currentOffset == null ||
!currentOffset.snapshotSplit()) {
return true;
}
}
@@ -752,7 +755,44 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
@Override
public String getPersistInfo() {
- return GsonUtils.GSON.toJson(this);
+ synchronized (splitsLock) {
+ return GsonUtils.GSON.toJson(this);
+ }
+ }
+
+ public String getPersistInfoWithoutTableSchemas() {
+ synchronized (splitsLock) {
+ JsonObject persisted =
GsonUtils.GSON.toJsonTree(this).getAsJsonObject();
+ persisted.remove(TABLE_SCHEMAS_FIELD);
+ return GsonUtils.GSON.toJson(persisted);
+ }
+ }
+
+ public void setTableSchemas(String tableSchemas) {
+ synchronized (splitsLock) {
+ this.tableSchemas = tableSchemas;
+ tableSchemasDirty = true;
+ }
+ }
+
+ // The job write lock spans payload creation, journal acknowledgement and
this call.
+ public void markTableSchemasPersisted() {
+ synchronized (splitsLock) {
+ tableSchemasDirty = false;
+ }
+ }
+
+ @Override
+ public String mergePersistInfo(String previousPersistInfo, String
incomingPersistInfo) {
+ JsonObject incoming = GsonUtils.GSON.fromJson(incomingPersistInfo,
JsonObject.class);
+ if (previousPersistInfo == null
+ || (incoming.has(TABLE_SCHEMAS_FIELD) &&
!incoming.get(TABLE_SCHEMAS_FIELD).isJsonNull())) {
+ return incomingPersistInfo;
+ }
+ // Missing schema means no update, just as in a CDC offset commit
request.
+ JsonObject previous = GsonUtils.GSON.fromJson(previousPersistInfo,
JsonObject.class);
+ incoming.add(TABLE_SCHEMAS_FIELD, previous.get(TABLE_SCHEMAS_FIELD));
+ return GsonUtils.GSON.toJson(incoming);
}
// ============ Async split progress (driven by scheduler each tick)
============
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
index 84cc3a94a53..0c5b72fccdc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
@@ -436,6 +436,12 @@ public class JdbcTvfSourceOffsetProvider extends
JdbcSourceOffsetProvider {
return remapped;
}
+ @Override
+ public String mergePersistInfo(String previousPersistInfo, String
incomingPersistInfo) {
+ // TVF records contain complete state rather than schema-omitting CDC
updates.
+ return incomingPersistInfo;
+ }
+
/**
* Restore chw/bop/ts from image or job update journal JSON; preserves
cumulative snapshot progress.
*/
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
index b3c6466e878..4530be5af9a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
@@ -2195,6 +2195,10 @@ public class EditLog {
logEdit(OperationType.OP_UPDATE_SCHEDULER_JOB, job);
}
+ public void logUpdateJob(String serializedJob) {
+ logEdit(OperationType.OP_UPDATE_SCHEDULER_JOB, out ->
Text.writeString(out, serializedJob));
+ }
+
public void logDeleteJob(AbstractJob job) {
logEdit(OperationType.OP_DELETE_SCHEDULER_JOB, job);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
index 0ffeb02f402..b0b30708725 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
@@ -505,6 +505,11 @@ public class StreamingInsertJobOffsetPersistenceTest {
private static class TestStreamingInsertJob extends StreamingInsertJob {
private int journalCount;
+ @Override
+ protected void logUpdateOperation(String persistInfo) {
+ journalCount++;
+ }
+
@Override
public void logUpdateOperation() {
journalCount++;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobSchemaPersistenceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobSchemaPersistenceTest.java
new file mode 100644
index 00000000000..01af8115493
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobSchemaPersistenceTest.java
@@ -0,0 +1,434 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.extensions.insert.streaming;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.CommitOffsetRequest;
+import org.apache.doris.job.cdc.split.SnapshotSplit;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.common.JobStatus;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.manager.JobManager;
+import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
+import org.apache.doris.journal.Journal;
+import org.apache.doris.journal.JournalEntity;
+import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.OperationType;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TxnStateCallbackFactory;
+
+import com.google.gson.JsonObject;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class StreamingInsertJobSchemaPersistenceTest {
+ private final List<byte[]> journals = new ArrayList<>();
+ private final List<Writable> payloads = new ArrayList<>();
+ private MockedStatic<Env> envMock;
+ private boolean oldBatchLog;
+ private boolean failNextWrite;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ oldBatchLog = Config.enable_batch_editlog;
+ Config.enable_batch_editlog = true;
+ envMock = Mockito.mockStatic(Env.class);
+ Env env = Mockito.mock(Env.class);
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ envMock.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+
Mockito.when(catalog.getDbOrAnalysisException("schema_persistence_test"))
+ .thenReturn(new Database(9002L, "schema_persistence_test"));
+ GlobalTransactionMgrIface transactions =
Mockito.mock(GlobalTransactionMgrIface.class);
+
envMock.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactions);
+ Mockito.when(transactions.getCallbackFactory()).thenReturn(new
TxnStateCallbackFactory());
+
+ EditLog editLog = Mockito.mock(EditLog.class,
Mockito.CALLS_REAL_METHODS);
+ Mockito.doReturn(1).when(editLog).getNumEditStreams();
+ Deencapsulation.setField(editLog, "numTransactions", new AtomicLong());
+ Deencapsulation.setField(editLog, "totalTimeTransactions", new
AtomicLong());
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Answer<Long> capture = invocation -> {
+ short op = invocation.getArgument(0);
+ Writable payload = invocation.getArgument(1);
+ if (failNextWrite) {
+ failNextWrite = false;
+ throw new IllegalStateException("injected journal failure");
+ }
+ // Exercise serialization on a different thread while commitOffset
still holds the job lock.
+ byte[] bytes = CompletableFuture.supplyAsync(() ->
encodeJournal(op, payload)).get(10, TimeUnit.SECONDS);
+ journals.add(bytes);
+ payloads.add(payload);
+ return (long) journals.size();
+ };
+
Mockito.doAnswer(capture).when(editLog).logEditWithQueue(Mockito.anyShort(),
Mockito.any(Writable.class));
+ Journal journal = Mockito.mock(Journal.class);
+ Mockito.doAnswer(capture).when(journal).write(Mockito.anyShort(),
Mockito.any(Writable.class));
+ Deencapsulation.setField(editLog, "journal", journal);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Config.enable_batch_editlog = oldBatchLog;
+ envMock.close();
+ }
+
+ @Test
+ public void testClientSchemaUpdatesAreWrittenAndNullIsOmitted() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ job.commitOffset(request("200", null));
+ job.commitOffset(request("300", "schema-A"));
+ job.commitOffset(request("400", "schema-B"));
+ job.commitOffset(request("500", null));
+
+ Assertions.assertEquals("schema-A", loggedSchema(0));
+ Assertions.assertNull(loggedSchema(1));
+ Assertions.assertEquals("schema-A", loggedSchema(2));
+ Assertions.assertEquals("schema-B", loggedSchema(3));
+ Assertions.assertNull(loggedSchema(4));
+ Assertions.assertEquals("schema-B", ((JdbcSourceOffsetProvider)
job.offsetProvider).getTableSchemas());
+ Assertions.assertArrayEquals(journals.get(0),
+ encodeJournal(OperationType.OP_UPDATE_SCHEDULER_JOB,
payloads.get(0)));
+ }
+
+ @Test
+ public void testFullJournalEncodingMatchesJobWriter() throws Exception {
+ StreamingInsertJob job = newJob();
+ ((JdbcSourceOffsetProvider)
job.offsetProvider).setTableSchemas("schema-A");
+ job.setOffsetProviderPersist(job.offsetProvider.getPersistInfo());
+ byte[] expected = encodeJournal(OperationType.OP_UPDATE_SCHEDULER_JOB,
job);
+
+ job.logUpdateOperation(job.getOffsetProviderPersist());
+
+ Assertions.assertArrayEquals(expected, journals.get(0));
+ }
+
+ @Test
+ public void testCheckpointAndFailoverRetainSchemaWithoutScheduling()
throws Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ job.commitOffset(request("200", null));
+ job.commitOffset(request("300", "schema-B"));
+ job.commitOffset(request("400", null));
+ JobManager<StreamingInsertJob, Map<Object, Object>> manager = new
JobManager<>();
+ Map<Long, StreamingInsertJob> jobs = Deencapsulation.getField(manager,
"jobMap");
+ jobs.put(9001L, newJob());
+ for (byte[] journal : journals) {
+ manager.replayUpdateJob(readJournal(journal));
+ }
+
+ ByteArrayOutputStream image = new ByteArrayOutputStream();
+ manager.write(new DataOutputStream(image));
+ JobManager<StreamingInsertJob, Map<Object, Object>> restored = new
JobManager<>();
+ restored.readFields(new DataInputStream(new
ByteArrayInputStream(image.toByteArray())));
+ StreamingInsertJob recovered = restored.getJob(9001L);
+ assertState(recovered, "schema-B", "400");
+
+ job.commitOffset(request("500", null));
+ restored.replayUpdateJob(readJournal(journals.get(4)));
+ assertState(recovered, "schema-B", "500");
+ recovered.replayOffsetProviderIfNeed();
+ Assertions.assertEquals("schema-B", ((JdbcSourceOffsetProvider)
recovered.offsetProvider).getTableSchemas());
+ assertState(recovered, "schema-B", "500");
+ Deencapsulation.setField(recovered, "runningStreamTask", new
NoopTask());
+ recovered.commitOffset(request("600", null));
+ Assertions.assertNull(loggedSchema(5));
+ }
+
+ @Test
+ public void testFailureRetainsSchemaForSameSchemaAndNullRetries() throws
Exception {
+ StreamingInsertJob job = newJob();
+ for (String retrySchema : new String[] {"schema-A", null}) {
+ failNextWrite = true;
+ String schema = retrySchema == null ? "schema-B" : "schema-A";
+ Assertions.assertThrows(IllegalStateException.class, () ->
job.commitOffset(request("100", schema)));
+ job.commitOffset(request("200", retrySchema));
+ Assertions.assertEquals(schema, loggedSchema(journals.size() - 1));
+ job.commitOffset(request("300", null));
+ Assertions.assertNull(loggedSchema(journals.size() - 1));
+ }
+ }
+
+ @Test
+ public void testSchemaChangeBypassesSnapshotThrottle() throws Exception {
+ StreamingInsertJob job = newJob();
+ JdbcSourceOffsetProvider provider = (JdbcSourceOffsetProvider)
job.offsetProvider;
+ for (int i = 0; i < 3; i++) {
+ String splitId = "table:" + i;
+ provider.getRemainingSplits().add(new SnapshotSplit(splitId,
"db.table",
+ Collections.singletonList("id"), new Object[] {1L}, new
Object[] {2L}, null));
+ CommitOffsetRequest request = request("100", i == 1 ? "schema-A" :
null);
+ request.setOffset("[{\"splitId\":\"" + splitId +
"\",\"lsn\":\"100\"}]");
+ job.commitOffset(request);
+ }
+ Assertions.assertEquals(2, journals.size());
+ Assertions.assertEquals("schema-A", loggedSchema(1));
+ }
+
+ @Test
+ public void testOrdinaryJournalSizeDoesNotGrowWithSchema() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "x".repeat(1024)));
+ job.commitOffset(request("200", null));
+ int smallJournalSize = journals.get(1).length;
+ job.commitOffset(request("300", "x".repeat(1024 * 1024)));
+ job.commitOffset(request("400", null));
+ Assertions.assertTrue(journals.get(2).length > 1024 * 1024);
+ Assertions.assertTrue(journals.get(3).length < 10_000);
+ Assertions.assertTrue(Math.abs(smallJournalSize -
journals.get(3).length) < 100);
+ }
+
+ @Test
+ public void testFullJobStateSurvivesCompactOffsetJournal() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+
+ job.commitOffset(request("200", null));
+ StreamingTaskTxnCommitAttachment attachment = new
StreamingTaskTxnCommitAttachment(
+ 9001L, 1001L, 0, 0, 0, 0, 0, request("300", null).getOffset());
+ Deencapsulation.invoke(job, "updateCloudJobStatisticAndOffset",
attachment, true);
+ job.logUpdateOperation();
+
+ Assertions.assertNull(loggedSchema(1));
+ Assertions.assertEquals("schema-A", loggedSchema(2));
+ assertState(job, "schema-A", "300");
+ assertState(GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(job),
StreamingInsertJob.class),
+ "schema-A", "300");
+ }
+
+ @Test
+ public void testCloudOffsetRecoveryBeforeProviderRestoreKeepsSchema()
throws Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ StreamingInsertJob recovered = readJournal(journals.get(0));
+ Assertions.assertNull(((JdbcSourceOffsetProvider)
recovered.offsetProvider).getTableSchemas());
+
+ StreamingTaskTxnCommitAttachment attachment = new
StreamingTaskTxnCommitAttachment(
+ 9001L, 1001L, 0, 0, 0, 0, 0, request("200", null).getOffset());
+ Deencapsulation.invoke(recovered, "updateCloudJobStatisticAndOffset",
attachment, true);
+ assertState(recovered, "schema-A", "200");
+ recovered.replayOffsetProviderIfNeed();
+ Assertions.assertEquals("schema-A", ((JdbcSourceOffsetProvider)
recovered.offsetProvider).getTableSchemas());
+ }
+
+ @Test
+ public void testDirectJournalOmitsUnchangedSchema() throws Exception {
+ Config.enable_batch_editlog = false;
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ job.commitOffset(request("200", null));
+ Assertions.assertEquals("schema-A", loggedSchema(0));
+ Assertions.assertNull(loggedSchema(1));
+ JsonObject persisted =
GsonUtils.GSON.fromJson(loggedJob(1).get("opp").getAsString(),
JsonObject.class);
+ Assertions.assertFalse(persisted.has("ts"));
+ StreamingInsertJob recovered = readJournal(journals.get(0));
+ recovered.replayOnUpdated(readJournal(journals.get(1)));
+ assertState(recovered, "schema-A", "200");
+ }
+
+ @Test
+ public void testRecordWithoutSchemaRetainsPreviousSchema() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ StreamingInsertJob recovered = readJournal(journals.get(0));
+ newJob().commitOffset(request("200", null));
+ recovered.replayOnUpdated(readJournal(journals.get(1)));
+ assertState(recovered, "schema-A", "200");
+ }
+
+ @Test
+ public void testFullImageAndAlterOffsetKeepSchema() throws Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ job.commitOffset(request("200", null));
+ job.setJobStatus(JobStatus.PAUSED);
+ job.alterJob(new AlterJobCommand(job.getJobName(),
+ Map.of(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"300\"}"),
+ null, null, null, Collections.emptyMap(),
Collections.emptyMap()));
+ ByteArrayOutputStream image = new ByteArrayOutputStream();
+ job.write(new DataOutputStream(image));
+ StreamingInsertJob restored = (StreamingInsertJob)
AbstractJob.readFields(
+ new DataInputStream(new
ByteArrayInputStream(image.toByteArray())));
+ assertState(restored, "schema-A", "300");
+ StreamingInsertJob follower = readJournal(journals.get(0));
+ follower.replayOnUpdated(readJournal(journals.get(2)));
+ assertState(follower, "schema-A", "300");
+ }
+
+ @Test
+ public void testAlterPausedJobAfterImageRecoveryKeepsSchema() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ job.setJobStatus(JobStatus.PAUSED);
+ ByteArrayOutputStream image = new ByteArrayOutputStream();
+ job.write(new DataOutputStream(image));
+ StreamingInsertJob recovered = (StreamingInsertJob)
AbstractJob.readFields(
+ new DataInputStream(new
ByteArrayInputStream(image.toByteArray())));
+
+ recovered.alterJob(new AlterJobCommand(recovered.getJobName(),
+ Map.of(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"200\"}"),
+ null, null, null, Collections.emptyMap(),
Collections.emptyMap()));
+ assertState(recovered, "schema-A", "200");
+ image.reset();
+ recovered.write(new DataOutputStream(image));
+ assertState((StreamingInsertJob) AbstractJob.readFields(
+ new DataInputStream(new
ByteArrayInputStream(image.toByteArray()))), "schema-A", "200");
+
+ recovered.setJobStatus(JobStatus.RUNNING);
+ recovered.replayOffsetProviderIfNeed();
+ Assertions.assertEquals("schema-A", ((JdbcSourceOffsetProvider)
recovered.offsetProvider).getTableSchemas());
+ Deencapsulation.setField(recovered, "runningStreamTask", new
NoopTask());
+ recovered.commitOffset(request("300", null));
+ assertState(recovered, "schema-A", "300");
+ }
+
+ @Test
+ public void testLegacyOffsetJournalRetainsPreviousSchema() throws
Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ StreamingInsertJob recovered = readJournal(journals.get(0));
+ StreamingInsertJob legacyUpdate = newJob();
+ legacyUpdate.setJobStatus(JobStatus.PAUSED);
+
legacyUpdate.getProperties().put(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"200\"}");
+
+ recovered.replayOnUpdated(legacyUpdate);
+
+ assertState(recovered, "schema-A", "200");
+ }
+
+ @Test
+ public void testStatusJournalAfterRecoveryRetainsSchemaWithoutScheduling()
throws Exception {
+ StreamingInsertJob job = newJob();
+ job.commitOffset(request("100", "schema-A"));
+ StreamingInsertJob recovered = readJournal(journals.get(0));
+ recovered.setJobStatus(JobStatus.PAUSED);
+ recovered.logUpdateOperation();
+ Assertions.assertEquals("schema-A", loggedSchema(1));
+
+ StreamingInsertJob follower = readJournal(journals.get(0));
+ follower.replayOnUpdated(readJournal(journals.get(1)));
+ assertState(follower, "schema-A", "100");
+ ByteArrayOutputStream image = new ByteArrayOutputStream();
+ follower.write(new DataOutputStream(image));
+ assertState((StreamingInsertJob) AbstractJob.readFields(
+ new DataInputStream(new
ByteArrayInputStream(image.toByteArray()))), "schema-A", "100");
+ }
+
+ private static StreamingInsertJob newJob() {
+ StreamingInsertJob job = new StreamingInsertJob();
+ job.setJobId(9001L);
+ job.setJobName("schema_persistence_test");
+ job.setCurrentDbName("schema_persistence_test");
+ job.setJobStatus(JobStatus.RUNNING);
+ Deencapsulation.setField(job, "dataSourceType",
DataSourceType.POSTGRES);
+ Deencapsulation.setField(job, "properties", new HashMap<String,
String>());
+ Map<String, String> sourceProperties = new HashMap<>(
+ Map.of(DataSourceConfigKeys.JDBC_URL,
"jdbc:postgresql://127.0.0.1:5432/source_db"));
+ Deencapsulation.setField(job, "sourceProperties", sourceProperties);
+ Deencapsulation.setField(job, "targetProperties", new HashMap<String,
String>());
+ Deencapsulation.setField(job, "runningStreamTask", new NoopTask());
+ job.offsetProvider = new JdbcSourceOffsetProvider(9001L,
DataSourceType.POSTGRES, sourceProperties);
+ return job;
+ }
+
+ private static CommitOffsetRequest request(String lsn, String schema) {
+ CommitOffsetRequest request = new CommitOffsetRequest();
+ request.setTaskId(1001L);
+ request.setOffset("[{\"splitId\":\"binlog-split\",\"lsn\":\"" + lsn +
"\"}]");
+ request.setTableSchemas(schema);
+ return request;
+ }
+
+ private static byte[] encodeJournal(short op, Writable payload) {
+ try {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeShort(op);
+ payload.write(out);
+ return bytes.toByteArray();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static StreamingInsertJob readJournal(byte[] bytes) throws
Exception {
+ JournalEntity entity = new JournalEntity();
+ entity.readFields(new DataInputStream(new
ByteArrayInputStream(bytes)));
+ Assertions.assertEquals(OperationType.OP_UPDATE_SCHEDULER_JOB,
entity.getOpCode());
+ return (StreamingInsertJob) entity.getData();
+ }
+
+ private JsonObject loggedJob(int index) throws Exception {
+ DataInputStream in = new DataInputStream(new
ByteArrayInputStream(journals.get(index)));
+ Assertions.assertEquals(OperationType.OP_UPDATE_SCHEDULER_JOB,
in.readShort());
+ return GsonUtils.GSON.fromJson(Text.readString(in), JsonObject.class);
+ }
+
+ private String loggedSchema(int index) throws Exception {
+ JsonObject offset =
GsonUtils.GSON.fromJson(loggedJob(index).get("opp").getAsString(),
JsonObject.class);
+ return offset.has("ts") ? offset.get("ts").getAsString() : null;
+ }
+
+ private static void assertState(StreamingInsertJob job, String schema,
String lsn) {
+ JdbcSourceOffsetProvider state = GsonUtils.GSON.fromJson(
+ job.getOffsetProviderPersist(),
JdbcSourceOffsetProvider.class);
+ Assertions.assertEquals(schema, state.getTableSchemas());
+ Assertions.assertEquals(lsn,
state.getBinlogOffsetPersist().get("lsn"));
+ }
+
+ private static class NoopTask extends StreamingMultiTblTask {
+ NoopTask() {
+ super(9001L, 1001L, null, null, null, null, null,
+ new StreamingJobProperties(new HashMap<>()), null, null);
+ Deencapsulation.setField(this, "status", TaskStatus.RUNNING);
+ }
+
+ @Override
+ public void successCallback(CommitOffsetRequest offsetRequest) {
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
index 8adac456c01..4c6b54dc1e0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
@@ -25,7 +25,10 @@ import org.apache.doris.job.cdc.split.BinlogSplit;
import org.apache.doris.job.cdc.split.SnapshotSplit;
import org.apache.doris.job.common.DataSourceType;
import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.persist.gson.GsonUtils;
+import com.google.gson.JsonObject;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -145,6 +148,82 @@ public class JdbcSourceOffsetProviderOffsetTest {
Assertions.assertTrue(provider.shouldPersistOffset(1_000L, 1_001L));
}
+ @Test
+ public void testSchemaRemainsPendingUntilJournalAcknowledgement() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ provider.setTableSchemas("schema-A");
+ Assertions.assertEquals("schema-A", GsonUtils.GSON.fromJson(
+ provider.getPersistInfo(),
JdbcSourceOffsetProvider.class).getTableSchemas());
+ // Preparing a record, or receiving the same schema again, must not
acknowledge it.
+ provider.setTableSchemas("schema-A");
+ Assertions.assertEquals("schema-A", GsonUtils.GSON.fromJson(
+ provider.getPersistInfo(),
JdbcSourceOffsetProvider.class).getTableSchemas());
+ Assertions.assertTrue(provider.isTableSchemasDirty());
+
+ provider.markTableSchemasPersisted();
+ JsonObject delta =
GsonUtils.GSON.fromJson(provider.getPersistInfoWithoutTableSchemas(),
JsonObject.class);
+ Assertions.assertFalse(delta.has("ts"));
+ Assertions.assertFalse(provider.isTableSchemasDirty());
+ Assertions.assertEquals("schema-A", provider.getTableSchemas());
+
+ provider.setTableSchemas("schema-A");
+ Assertions.assertTrue(provider.isTableSchemasDirty());
+ provider.setTableSchemas("schema-B");
+ Assertions.assertEquals("schema-B", GsonUtils.GSON.fromJson(
+ provider.getPersistInfo(),
JdbcSourceOffsetProvider.class).getTableSchemas());
+ }
+
+ @Test
+ public void testSchemaFreePersistInfoRetainsOffsetAndSplitProgress() {
+ JdbcSourceOffsetProvider provider = GsonUtils.GSON.fromJson(
+ "{\"chw\":{\"db.table\":{\"db.table:0\":{\"lsn\":\"90\"}}},"
+ +
"\"bop\":{\"lsn\":\"100\"},\"ts\":\"schema-A\",\"csp\":{\"ni\":2}}",
+ JdbcSourceOffsetProvider.class);
+ String fullState = provider.getPersistInfo();
+
+ JsonObject persisted =
GsonUtils.GSON.fromJson(provider.getPersistInfoWithoutTableSchemas(),
JsonObject.class);
+
+ Assertions.assertFalse(persisted.has("ts"));
+ Assertions.assertEquals("100",
persisted.getAsJsonObject("bop").get("lsn").getAsString());
+ Assertions.assertEquals("90",
persisted.getAsJsonObject("chw").getAsJsonObject("db.table")
+ .getAsJsonObject("db.table:0").get("lsn").getAsString());
+ Assertions.assertEquals(2,
persisted.getAsJsonObject("csp").get("ni").getAsInt());
+ Assertions.assertEquals(fullState, provider.getPersistInfo());
+ }
+
+ @Test
+ public void testPersistedStateMergesWithoutRuntimeInitialization() {
+ SourceOffsetProvider merger = new JdbcSourceOffsetProvider();
+ String previous = "{\"bop\":{\"lsn\":\"100\"},\"ts\":\"schema-A\"}";
+ String merged = merger.mergePersistInfo(previous,
"{\"bop\":{\"lsn\":\"200\"}}");
+ JdbcSourceOffsetProvider provider = GsonUtils.GSON.fromJson(merged,
JdbcSourceOffsetProvider.class);
+ Assertions.assertEquals("schema-A", provider.getTableSchemas());
+ Assertions.assertEquals("200",
provider.getBinlogOffsetPersist().get("lsn"));
+ Assertions.assertNull(provider.getCurrentOffset());
+
+ String explicitNull = merger.mergePersistInfo(merged,
+ "{\"bop\":{\"lsn\":\"300\"},\"ts\":null}");
+ JdbcSourceOffsetProvider recovered =
GsonUtils.GSON.fromJson(explicitNull, JdbcSourceOffsetProvider.class);
+ Assertions.assertEquals("schema-A", recovered.getTableSchemas());
+ Assertions.assertEquals("300",
recovered.getBinlogOffsetPersist().get("lsn"));
+ String updated = "{\"bop\":{\"lsn\":\"400\"},\"ts\":\"schema-B\"}";
+ Assertions.assertEquals(updated, merger.mergePersistInfo(explicitNull,
updated));
+ Assertions.assertEquals(updated, merger.mergePersistInfo(null,
updated));
+ }
+
+ @Test
+ public void testTvfPersistedStateWithoutSchemaReplacesPreviousSchema() {
+ SourceOffsetProvider provider = new JdbcTvfSourceOffsetProvider();
+ String previous = "{\"bop\":{\"lsn\":\"100\"},\"ts\":\"schema-A\"}";
+ for (String incoming : new String[] {
+ "{\"bop\":{\"lsn\":\"200\"}}",
"{\"bop\":{\"lsn\":\"200\"},\"ts\":null}"}) {
+ JdbcSourceOffsetProvider recovered = GsonUtils.GSON.fromJson(
+ provider.mergePersistInfo(previous, incoming),
JdbcSourceOffsetProvider.class);
+ Assertions.assertNull(recovered.getTableSchemas());
+ Assertions.assertEquals("200",
recovered.getBinlogOffsetPersist().get("lsn"));
+ }
+ }
+
@Test
public void testEndOffsetAdvancesWhenCurrentOffsetIsAhead() {
assertEndOffsetAdvancesWhenCurrentOffsetIsAhead(new
TestJdbcSourceOffsetProvider(-1));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
