This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 a29db1c7308 [fix](fe) Propagate JSON properties for multi-table
routine load (#67510)
a29db1c7308 is described below
commit a29db1c730883c17a9b7a5ab4e9c3cf4e26ab400
Author: Refrain <[email protected]>
AuthorDate: Wed Oct 21 10:51:18 2026 +0800
[fix](fe) Propagate JSON properties for multi-table routine load (#67510)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Multi-table routine load cannot attach a table-specific
pipeline plan before the backend discovers the destination table. The
backend therefore requests a per-table plan with the format, and FE
rebuilds the stream-load task from the Routine Load job. That rebuild
already restored `json_root` and `strip_outer_array`, but omitted
`jsonpaths` and `num_as_string`. Nested JSON records were consequently
planned with an empty JSON path list and filtered as invalid rows. This
change restores those two properties before generating each
dynamic-table pipeline plan.
---
.../doris/nereids/load/NereidsStreamLoadTask.java | 2 ++
.../nereids/load/NereidsStreamLoadTaskTest.java | 31 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
index be92b7cde06..95e31360ced 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
@@ -383,6 +383,8 @@ public class NereidsStreamLoadTask implements
NereidsLoadTaskInfo {
this.timezone = task.getTimezone();
this.formatType = task.getFormatType();
this.stripOuterArray = task.isStripOuterArray();
+ this.numAsString = task.isNumAsString();
+ this.jsonPaths = task.getJsonPaths();
this.jsonRoot = task.getJsonRoot();
this.sendBatchParallelism = task.getSendBatchParallelism();
this.loadToSingleTablet = task.isLoadToSingleTablet();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
index 7e690bec0fa..403766a6769 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
@@ -18,6 +18,7 @@
package org.apache.doris.nereids.load;
import org.apache.doris.common.UserException;
+import org.apache.doris.task.LoadTaskInfo;
import org.apache.doris.thrift.TFileCompressType;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileType;
@@ -26,8 +27,38 @@ import org.apache.doris.thrift.TUniqueId;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
public class NereidsStreamLoadTaskTest {
+ @Test
+ public void testMultiTableBaseTaskCopiesJsonProperties() throws Exception {
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setLoadId(new TUniqueId(1, 2));
+ request.setTxnId(3);
+ request.setFileType(TFileType.FILE_STREAM);
+ request.setFormatType(TFileFormatType.FORMAT_JSON);
+ request.setCompressType(TFileCompressType.PLAIN);
+
+ NereidsStreamLoadTask streamLoadTask =
NereidsStreamLoadTask.fromTStreamLoadPutRequest(request);
+ LoadTaskInfo routineLoadTask = Mockito.mock(LoadTaskInfo.class);
+
Mockito.when(routineLoadTask.getFormatType()).thenReturn(TFileFormatType.FORMAT_JSON);
+ Mockito.when(routineLoadTask.getJsonPaths()).thenReturn(
+ "[\"$.meta.id\", \"$.meta.ts\", \"$.value.score\",
\"$.value.region\"]");
+
Mockito.when(routineLoadTask.getJsonRoot()).thenReturn("$.payload.items");
+ Mockito.when(routineLoadTask.isStripOuterArray()).thenReturn(true);
+ Mockito.when(routineLoadTask.isNumAsString()).thenReturn(true);
+
+ streamLoadTask.setMultiTableBaseTaskInfo(routineLoadTask);
+
+ Assertions.assertEquals(TFileFormatType.FORMAT_JSON,
streamLoadTask.getFormatType());
+ Assertions.assertEquals(
+ "[\"$.meta.id\", \"$.meta.ts\", \"$.value.score\",
\"$.value.region\"]",
+ streamLoadTask.getJsonPaths());
+ Assertions.assertEquals("$.payload.items",
streamLoadTask.getJsonRoot());
+ Assertions.assertTrue(streamLoadTask.isStripOuterArray());
+ Assertions.assertTrue(streamLoadTask.isNumAsString());
+ }
+
@Test
public void testDefaultSendBatchParallelism() throws UserException {
Assertions.assertEquals(1,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
