This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new bbc790ed70a branch-4.2: [fix](streaming-job) Avoid redundant schema
persistence in CDC offset journals #69129 (#69180)
bbc790ed70a is described below
commit bbc790ed70a300871428e3fda0e2b6b71e489090
Author: wudi <[email protected]>
AuthorDate: Fri Oct 16 23:28:16 2026 +0800
branch-4.2: [fix](streaming-job) Avoid redundant schema persistence in CDC
offset journals #69129 (#69180)
Cherry-picked from #69129
---
.../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 | 435 +++++++++++++++++++++
.../jdbc/JdbcSourceOffsetProviderOffsetTest.java | 79 ++++
8 files changed, 602 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 f5fb8b090a5..71f94723427 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());
@@ -1027,7 +1030,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) {
@@ -1102,6 +1107,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.
@@ -1114,8 +1123,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);
}
@@ -1124,7 +1133,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
@@ -1403,6 +1413,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();
@@ -1736,7 +1753,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 37e7136743e..e5ffe6ab25e 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;
}
}
@@ -758,7 +761,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 ea93b941cf1..c5450abd6ba 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
@@ -2107,6 +2107,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 84055842cb1..65d94c8e672 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
@@ -506,6 +506,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..3500d2ddd39
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobSchemaPersistenceTest.java
@@ -0,0 +1,435 @@
+// 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.common.base.Strings;
+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", Strings.repeat("x", 1024)));
+ job.commitOffset(request("200", null));
+ int smallJournalSize = journals.get(1).length;
+ job.commitOffset(request("300", Strings.repeat("x", 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(),
+
Collections.singletonMap(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(),
+
Collections.singletonMap(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<>(
+ Collections.singletonMap(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 1ea32f914a1..cbdaa98277e 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.Assert;
import org.junit.Test;
@@ -145,6 +148,82 @@ public class JdbcSourceOffsetProviderOffsetTest {
Assert.assertTrue(provider.shouldPersistOffset(1_000L, 1_001L));
}
+ @Test
+ public void testSchemaRemainsPendingUntilJournalAcknowledgement() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ provider.setTableSchemas("schema-A");
+ Assert.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");
+ Assert.assertEquals("schema-A", GsonUtils.GSON.fromJson(
+ provider.getPersistInfo(),
JdbcSourceOffsetProvider.class).getTableSchemas());
+ Assert.assertTrue(provider.isTableSchemasDirty());
+
+ provider.markTableSchemasPersisted();
+ JsonObject delta =
GsonUtils.GSON.fromJson(provider.getPersistInfoWithoutTableSchemas(),
JsonObject.class);
+ Assert.assertFalse(delta.has("ts"));
+ Assert.assertFalse(provider.isTableSchemasDirty());
+ Assert.assertEquals("schema-A", provider.getTableSchemas());
+
+ provider.setTableSchemas("schema-A");
+ Assert.assertTrue(provider.isTableSchemasDirty());
+ provider.setTableSchemas("schema-B");
+ Assert.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);
+
+ Assert.assertFalse(persisted.has("ts"));
+ Assert.assertEquals("100",
persisted.getAsJsonObject("bop").get("lsn").getAsString());
+ Assert.assertEquals("90",
persisted.getAsJsonObject("chw").getAsJsonObject("db.table")
+ .getAsJsonObject("db.table:0").get("lsn").getAsString());
+ Assert.assertEquals(2,
persisted.getAsJsonObject("csp").get("ni").getAsInt());
+ Assert.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);
+ Assert.assertEquals("schema-A", provider.getTableSchemas());
+ Assert.assertEquals("200",
provider.getBinlogOffsetPersist().get("lsn"));
+ Assert.assertNull(provider.getCurrentOffset());
+
+ String explicitNull = merger.mergePersistInfo(merged,
+ "{\"bop\":{\"lsn\":\"300\"},\"ts\":null}");
+ JdbcSourceOffsetProvider recovered =
GsonUtils.GSON.fromJson(explicitNull, JdbcSourceOffsetProvider.class);
+ Assert.assertEquals("schema-A", recovered.getTableSchemas());
+ Assert.assertEquals("300",
recovered.getBinlogOffsetPersist().get("lsn"));
+ String updated = "{\"bop\":{\"lsn\":\"400\"},\"ts\":\"schema-B\"}";
+ Assert.assertEquals(updated, merger.mergePersistInfo(explicitNull,
updated));
+ Assert.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);
+ Assert.assertNull(recovered.getTableSchemas());
+ Assert.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]
