Copilot commented on code in PR #69129:
URL: https://github.com/apache/doris/pull/69129#discussion_r4281091955
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -752,7 +755,44 @@ private static String getTableName(String tableId) {
@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));
Review Comment:
`mergePersistInfo` can throw at runtime when the previous payload does not
contain `"ts"` (or contains it as absent), because
`previous.get(TABLE_SCHEMAS_FIELD)` may return `null` and `JsonObject#add(...)`
does not accept a null `JsonElement`. It also re-introduces `"ts": null` into
an otherwise compact record if the previous value is JSON null. A concrete fix
is to only merge the schema field when `previous.has("ts")` and the value is
non-null (`!isJsonNull()`); otherwise leave it absent in `incoming`.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1427,6 +1437,13 @@ public void write(DataOutput out) throws IOException {
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);
Review Comment:
The `"opp"` field name is introduced as a hard-coded string here, while
other payload keys (e.g., `"ts"`) were moved to constants. To avoid drift
between serialization annotations and replay logic, extract `"opp"` into a
shared constant (ideally reusing the same constant used by the
`@SerializedName` for offset-provider persist, if present) and reference it
here.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -752,7 +755,44 @@ private static String getTableName(String tableId) {
@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);
+ }
+ }
Review Comment:
`getPersistInfoWithoutTableSchemas()` builds a full JSON tree for the entire
provider state and then serializes it again, which adds avoidable CPU and
allocations on the hot path of frequent offset journaling. Consider writing a
compact JSON form via a purpose-built writer (e.g., Gson streaming `JsonWriter`
/ custom adapter) that emits all persisted fields except `"ts"` without
constructing an intermediate `JsonObject`, or caching the schema-free
serialized form when only offsets/progress change.
##########
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java:
##########
@@ -2195,6 +2195,10 @@ public void logUpdateJob(AbstractJob job) {
logEdit(OperationType.OP_UPDATE_SCHEDULER_JOB, job);
}
+ public void logUpdateJob(String serializedJob) {
+ logEdit(OperationType.OP_UPDATE_SCHEDULER_JOB, out ->
Text.writeString(out, serializedJob));
Review Comment:
This overload takes an untyped `String` whose required format is “the exact
serialized payload expected by OP_UPDATE_SCHEDULER_JOB replay” (JSON string via
`Text.writeString`). To reduce misuse risk, consider renaming it to something
format-revealing (e.g., `logUpdateJobSerializedJson(...)`) and/or documenting
explicitly that the content must match the existing journal encoding for
scheduler-job updates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
