This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12305-befdd365521c3226fab0b8610dad28bbcaad4371 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 3e5bbfaa14bee630b5a60ad9887c304d48c544db Author: Goutam Adwant <[email protected]> AuthorDate: Sat Oct 24 15:46:19 2026 +0000 [Fix][API] Correct nested row byte accounting (#12305) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/introduction/configuration/speed-limit.md | 4 + docs/zh/introduction/configuration/speed-limit.md | 4 + .../seatunnel/api/table/type/SeaTunnelRow.java | 6 + .../seatunnel/api/table/type/SeaTunnelRowTest.java | 90 ++++++++++++++ .../server/master/NestedRowAccountingTest.java | 56 +++++++++ .../SeaTunnelSourceCollectorRowAccountingTest.java | 136 +++++++++++++++++++++ .../resources/batch_nested_rows_to_console.conf | 46 +++++++ .../format/json/JsonRowDataSerDeSchemaTest.java | 48 ++++++++ 8 files changed, 390 insertions(+) diff --git a/docs/en/introduction/configuration/speed-limit.md b/docs/en/introduction/configuration/speed-limit.md index 87379e5b75..59f6c5082e 100644 --- a/docs/en/introduction/configuration/speed-limit.md +++ b/docs/en/introduction/configuration/speed-limit.md @@ -7,6 +7,10 @@ This functionality is essential when you need to ensure efficient and controlled The speed control is primarily governed by two key parameters: `read_limit.rows_per_second` and `read_limit.bytes_per_second`. This document will guide you through the usage of these parameters and how to leverage them effectively. +In Zeta, the byte limit uses SeaTunnel's row-size estimate, not the exact serialized or network byte count. +For arrays of maps or rows, null elements contribute zero and non-null elements contribute their contents. +Correcting previously omitted row-array contents can increase reported byte metrics and reduce throughput under the same byte limit. + ## Support Those Engines > SeaTunnel Zeta<br/> diff --git a/docs/zh/introduction/configuration/speed-limit.md b/docs/zh/introduction/configuration/speed-limit.md index 51007269dd..5e0f6f02a0 100644 --- a/docs/zh/introduction/configuration/speed-limit.md +++ b/docs/zh/introduction/configuration/speed-limit.md @@ -6,6 +6,10 @@ SeaTunnel提供了强大的速度控制功能允许你管理数据同步的速 速度控制主要由两个关键参数控制：`read_limit.rows_per_second` 和 `read_limit.bytes_per_second`。 本文档将指导您如何使用这些参数以及如何有效地利用它们。 +在 Zeta 中，字节限速使用 SeaTunnel 的行大小估算值，而不是精确的序列化大小或网络传输字节数。 +对于 Map 数组或 Row 数组，null 元素按零计算，非 null 元素按其内容计算。 +修正此前遗漏的 Row 数组内容后，报告的字节指标可能增大，在相同字节限速下吞吐量可能降低。 + ## 支持这些引擎 > SeaTunnel Zeta<br/> diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java index e6b9d545a7..3acb5b2ed7 100644 --- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java +++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java @@ -260,6 +260,8 @@ public final class SeaTunnelRow implements Serializable { case DOUBLE: return getArrayNotNullSize((Double[]) v) * 8; case ARRAY: + case ROW: + // Nested elements need their declared type to account for their contents. int total = 0; for (Object elem : (Object[]) v) { total += getBytesForValue(elem, dataType); @@ -287,6 +289,10 @@ public final class SeaTunnelRow implements Serializable { int size = 0; if (Objects.nonNull(v)) { for (Map o : (Map[]) v) { + if (o == null) { + // Null array elements contribute no payload bytes, like other nullable values. + continue; + } for (Map.Entry<?, ?> entry : ((Map<?, ?>) o).entrySet()) { size += getBytesForValue(entry.getKey()) + getBytesForValue(entry.getValue()); } diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/SeaTunnelRowTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/SeaTunnelRowTest.java index 8dcb8b2672..3453c4e469 100644 --- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/SeaTunnelRowTest.java +++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/table/type/SeaTunnelRowTest.java @@ -242,6 +242,96 @@ public class SeaTunnelRowTest { Assertions.assertEquals(8, row.getBytesSize()); } + @Test + void testNullableMapArraySize() { + Map<String, Integer> map = new LinkedHashMap<>(); + map.put("a", 1); + map.put("b", null); + map.put(null, 2); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"items"}, + new SeaTunnelDataType<?>[] { + new ArrayType<>( + Map[].class, + new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE)) + }); + + assertSizeInBothOrders( + new Object[] {new Map[] {null, map, null, Collections.singletonMap("c", 3), null}}, + rowType, + 15); + assertSizeInBothOrders(new Object[] {new Map[] {null, null}}, rowType, 0); + assertSizeInBothOrders(new Object[] {new Map[0]}, rowType, 0); + assertSizeInBothOrders(new Object[] {null}, rowType, 0); + } + + @Test + void testNullableRowArraySize() { + SeaTunnelRowType elementType = + new SeaTunnelRowType( + new String[] {"value"}, new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE}); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "items"}, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, new ArrayType<>(SeaTunnelRow[].class, elementType) + }); + + assertSizeInBothOrders( + new Object[] { + 7, + new SeaTunnelRow[] { + null, + new SeaTunnelRow(new Object[] {"abcd"}), + new SeaTunnelRow(new Object[] {null}), + null + } + }, + rowType, + 8); + assertSizeInBothOrders(new Object[] {7, new SeaTunnelRow[] {null}}, rowType, 4); + assertSizeInBothOrders(new Object[] {7, new SeaTunnelRow[0]}, rowType, 4); + assertSizeInBothOrders(new Object[] {7, null}, rowType, 4); + } + + @Test + void testNestedRowArraySize() { + SeaTunnelRowType elementType = + new SeaTunnelRowType( + new String[] {"items"}, + new SeaTunnelDataType<?>[] { + new ArrayType<>( + Map[].class, + new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE)) + }); + ArrayType<?, ?> rowArrayType = new ArrayType<>(SeaTunnelRow[].class, elementType); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"groups"}, + new SeaTunnelDataType<?>[] { + new ArrayType<>(SeaTunnelRow[][].class, rowArrayType) + }); + SeaTunnelRow element = + new SeaTunnelRow(new Object[] {new Map[] {null, Collections.singletonMap("a", 1)}}); + + assertSizeInBothOrders( + new Object[] {new SeaTunnelRow[][] {null, new SeaTunnelRow[] {null, element}}}, + rowType, + 5); + } + + private static void assertSizeInBothOrders( + Object[] fields, SeaTunnelRowType rowType, int expectedSize) { + SeaTunnelRow typedFirst = new SeaTunnelRow(fields); + Assertions.assertEquals(expectedSize, typedFirst.getBytesSize(rowType)); + Assertions.assertEquals(expectedSize, typedFirst.getBytesSize()); + SeaTunnelRow untypedFirst = new SeaTunnelRow(fields); + Assertions.assertEquals(expectedSize, untypedFirst.getBytesSize()); + Assertions.assertEquals(expectedSize, untypedFirst.getBytesSize(rowType)); + Assertions.assertSame(fields, typedFirst.getFields()); + } + @Test void testWithMapInterface() { Map<String, String> map = Collections.singletonMap("key", "value"); diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/NestedRowAccountingTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/NestedRowAccountingTest.java new file mode 100644 index 0000000000..d2d5d29668 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/NestedRowAccountingTest.java @@ -0,0 +1,56 @@ +/* + * 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.seatunnel.engine.server.master; + +import org.apache.seatunnel.api.common.metrics.JobMetrics; +import org.apache.seatunnel.engine.common.job.JobStatus; +import org.apache.seatunnel.engine.server.AbstractSeaTunnelServerTest; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; + +import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES; +import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_COUNT; +import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES; +import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_COUNT; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; + +class NestedRowAccountingTest extends AbstractSeaTunnelServerTest<NestedRowAccountingTest> { + + @Test + void testNullableMapArraysReachSinkWithCorrectByteMetrics() { + long jobId = System.currentTimeMillis(); + startJob(jobId, "batch_nested_rows_to_console.conf", false); + + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> { + assertEquals( + JobStatus.FINISHED, + server.getCoordinatorService().getJobStatus(jobId)); + JobMetrics metrics = + server.getCoordinatorService().getJobMetrics(jobId); + assertEquals(3L, metrics.get(SOURCE_RECEIVED_COUNT).get(0).value()); + assertEquals(3L, metrics.get(SINK_WRITE_COUNT).get(0).value()); + assertEquals(17L, metrics.get(SOURCE_RECEIVED_BYTES).get(0).value()); + assertEquals(17L, metrics.get(SINK_WRITE_BYTES).get(0).value()); + }); + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorRowAccountingTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorRowAccountingTest.java new file mode 100644 index 0000000000..a8ee41c8e1 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollectorRowAccountingTest.java @@ -0,0 +1,136 @@ +/* + * 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.seatunnel.engine.server.task; + +import org.apache.seatunnel.api.common.metrics.Counter; +import org.apache.seatunnel.api.common.metrics.Meter; +import org.apache.seatunnel.api.common.metrics.MetricsContext; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.MultipleRowType; +import org.apache.seatunnel.api.table.type.Record; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.core.starter.flowcontrol.FlowControlStrategy; +import org.apache.seatunnel.engine.common.config.EngineConfig; +import org.apache.seatunnel.engine.server.execution.TaskGroupLocation; +import org.apache.seatunnel.engine.server.execution.TaskLocation; +import org.apache.seatunnel.engine.server.task.flow.OneInputFlowLifeCycle; + +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Map; + +import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES; +import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_COUNT; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class SeaTunnelSourceCollectorRowAccountingTest { + + @Test + void testNullableMapArrayCollection() throws Exception { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"items"}, + new SeaTunnelDataType<?>[] { + new ArrayType<>( + Map[].class, + new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE)) + }); + assertCollected( + new Object[] {new Map[] {null, Collections.singletonMap("a", 1)}}, rowType, 5); + assertCollected(new Object[] {new Map[] {null}}, rowType, 0); + } + + @Test + void testRowArrayCollection() throws Exception { + SeaTunnelRowType elementType = + new SeaTunnelRowType( + new String[] {"value"}, new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE}); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "items"}, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, new ArrayType<>(SeaTunnelRow[].class, elementType) + }); + assertCollected( + new Object[] { + 7, new SeaTunnelRow[] {null, new SeaTunnelRow(new Object[] {"abcd"})} + }, + rowType, + 8); + } + + private void assertCollected(Object[] fields, SeaTunnelRowType rowType, long expectedBytes) + throws Exception { + TablePath tablePath = TablePath.of("database", "table"); + for (boolean multiTable : new boolean[] {false, true}) { + SeaTunnelRow row = new SeaTunnelRow(fields); + row.setTableId(tablePath.getFullName()); + MetricsContext metrics = Mockito.mock(MetricsContext.class); + when(metrics.counter(Mockito.anyString())) + .thenAnswer(invocation -> Mockito.mock(Counter.class)); + when(metrics.meter(Mockito.anyString())) + .thenAnswer(invocation -> Mockito.mock(Meter.class)); + Counter bytes = Mockito.mock(Counter.class); + Counter count = Mockito.mock(Counter.class); + Counter tableBytes = Mockito.mock(Counter.class); + when(metrics.counter(SOURCE_RECEIVED_BYTES)).thenReturn(bytes); + when(metrics.counter(SOURCE_RECEIVED_COUNT)).thenReturn(count); + when(metrics.counter(SOURCE_RECEIVED_BYTES + "#" + tablePath.getFullName())) + .thenReturn(tableBytes); + OneInputFlowLifeCycle<Record<?>> output = Mockito.mock(OneInputFlowLifeCycle.class); + SeaTunnelTask sourceTask = Mockito.mock(SeaTunnelTask.class); + when(sourceTask.getTaskLocation()) + .thenReturn(new TaskLocation(new TaskGroupLocation(1L, 1, 1L), 1L, 0)); + SeaTunnelSourceCollector<SeaTunnelRow> collector = + new SeaTunnelSourceCollector<>( + new Object(), + Collections.singletonList(output), + metrics, + FlowControlStrategy.ofBytes(1000), + multiTable + ? new MultipleRowType( + Collections.singletonMap( + tablePath.getFullName(), rowType)) + : rowType, + multiTable + ? Collections.singletonList(tablePath) + : Collections.emptyList(), + sourceTask, + new EngineConfig(), + Collections.emptyMap(), + null); + + collector.collect(row); + + verify(bytes).inc(expectedBytes); + verify(count).inc(); + if (multiTable) { + verify(tableBytes).inc(expectedBytes); + } + verify(output).received(Mockito.argThat(record -> record.getData() == row)); + } + } +} diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/resources/batch_nested_rows_to_console.conf b/seatunnel-engine/seatunnel-engine-server/src/test/resources/batch_nested_rows_to_console.conf new file mode 100644 index 0000000000..f6703f276b --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/test/resources/batch_nested_rows_to_console.conf @@ -0,0 +1,46 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" + read_limit.bytes_per_second = 1000 +} + +source { + FakeSource { + row.num = 3 + split.num = 1 + schema = { + fields { + id = "int" + maps = "array<map<string,int>>" + } + } + rows = [ + { kind = INSERT, fields = { id = 1, maps = [null, {a = 1}] } }, + { kind = INSERT, fields = { id = 2, maps = [null] } }, + { kind = INSERT, fields = { id = 3, maps = [] } } + ] + } +} + +sink { + Console { + log.print.data = false + } +} diff --git a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java index b719b5b814..8e333dd777 100644 --- a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java +++ b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java @@ -98,6 +98,54 @@ public class JsonRowDataSerDeSchemaTest { new String[] {"id", "outer", "nested"}, new SeaTunnelDataType<?>[] {INT_TYPE, MAP_ARRAY_TYPE, ARRAY_MAP_TYPE}); + @Test + void testNullableMapArrayAccountingAfterDeserialization() throws Exception { + SeaTunnelRowType schema = + new SeaTunnelRowType( + new String[] {"items"}, + new SeaTunnelDataType<?>[] { + new ArrayType<>(Map[].class, new MapType<>(STRING_TYPE, INT_TYPE)) + }); + JsonDeserializationSchema deserializer = + new JsonDeserializationSchema( + CatalogTableUtil.getCatalogTable("", "", "", "test", schema), false, false); + byte[] json = "{\"items\":[null,{\"a\":1}]}".getBytes(StandardCharsets.UTF_8); + + SeaTunnelRow row = deserializer.deserialize(json); + Map<?, ?>[] items = (Map<?, ?>[]) row.getField(0); + assertNull(items[0]); + assertEquals(1, items[1].get("a")); + assertEquals(5, row.getBytesSize(schema)); + assertEquals(5, row.getBytesSize()); + assertEquals(5, deserializer.deserialize(json).getBytesSize()); + } + + @Test + void testRowArrayAccountingAfterDeserialization() throws Exception { + SeaTunnelRowType elementType = + new SeaTunnelRowType( + new String[] {"value"}, new SeaTunnelDataType<?>[] {STRING_TYPE}); + SeaTunnelRowType schema = + new SeaTunnelRowType( + new String[] {"id", "items"}, + new SeaTunnelDataType<?>[] { + INT_TYPE, new ArrayType<>(SeaTunnelRow[].class, elementType) + }); + JsonDeserializationSchema deserializer = + new JsonDeserializationSchema( + CatalogTableUtil.getCatalogTable("", "", "", "test", schema), false, false); + byte[] json = + "{\"id\":7,\"items\":[null,{\"value\":\"abcd\"}]}".getBytes(StandardCharsets.UTF_8); + + SeaTunnelRow row = deserializer.deserialize(json); + SeaTunnelRow[] items = (SeaTunnelRow[]) row.getField(1); + assertNull(items[0]); + assertEquals("abcd", items[1].getField(0)); + assertEquals(8, row.getBytesSize(schema)); + assertEquals(8, row.getBytesSize()); + assertEquals(8, deserializer.deserialize(json).getBytesSize()); + } + @Test public void testTimestampTzSerializedAsSessionWallClock() { // Issue #10795: with serializeTimestampTzAsLocal=true (used by the Doris
