From 93af3bec8a2cb95d89590f8ba601c61b1985d0cf Mon Sep 17 00:00:00 2001 From: Refrain Date: Fri, 4 Sep 2026 01:27:19 +0800 Subject: [PATCH] [fix](fe) Propagate JSON properties for multi-table routine load ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Multi-table routine load defers planning until the backend discovers each target table. The follow-up FE planning task restored json_root and strip_outer_array from the routine load job but omitted jsonpaths and num_as_string, so nested JSON records were planned without their configured paths and were filtered as invalid rows. Copy these properties into the per-table stream load task so the generated scan parameters match the routine load definition. ### Release note Fix multi-table routine load to honor jsonpaths and num_as_string for dynamic target tables. ### Check List (For Author) - Test: Unit Test and FE build - sh run-fe-ut.sh --run org.apache.doris.nereids.load.NereidsStreamLoadTaskTest - DISABLE_BUILD_UI=ON ./build.sh --fe -j48 - Behavior changed: Yes. Multi-table routine load now applies configured jsonpaths and num_as_string during per-table planning. - Does this need documentation: No --- .../nereids/load/NereidsStreamLoadTask.java | 2 + .../load/NereidsStreamLoadTaskTest.java | 60 +++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java 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 f5ddca41f19a29..654f749166b681 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 @@ -382,6 +382,8 @@ public void setMultiTableBaseTaskInfo(LoadTaskInfo task) throws UserException { 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 new file mode 100644 index 00000000000000..97591c68c0aa81 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java @@ -0,0 +1,60 @@ +// 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.nereids.load; + +import org.apache.doris.task.LoadTaskInfo; +import org.apache.doris.thrift.TFileCompressType; +import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TFileType; +import org.apache.doris.thrift.TStreamLoadPutRequest; +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()); + } +}