This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 ffe50311340 [fix](fe) Guard stream load schema metadata reads (#68964)
ffe50311340 is described below
commit ffe50311340b86b8340d035ef395f346113281c8
Author: Oliveira <[email protected]>
AuthorDate: Thu Oct 15 09:33:44 2026 +0800
[fix](fe) Guard stream load schema metadata reads (#68964)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68726
Problem Summary:
Stream Load reads the base schema version from table metadata outside
the planning lock. Concurrent schema change removes the old base-index
metadata before switching the base index ID under the table write lock.
An unlocked read in that interval can dereference missing metadata and
fail the load. A deterministic unit test reproduces the null dereference
on the pre-optimization baseline.
For regular Stream Load, populate the response database ID, table ID,
and base schema version inside the existing `generatePlan(OlapTable)`
read-lock scope. Populate them only for the first planned table when a
response is present, preserving multi-table behavior. Planning and
response metadata use the same lock scope without an additional lock
acquisition.
For HTTP `load_sql`, obtain the response table ID and base schema
version from the generated `OlapTableSink`. These values were already
captured during planning under the existing table lock. Keep the other
table-property reads as before. This removes the second metadata-lock
acquisition and the transaction-abort branch introduced solely for its
timeout. Existing transaction handling and Thrift fields are preserved.
This fix is separate from the scheduling and file TVF COUNT optimization
in #68726.
### Release note
Prevent Stream Load response schema-version reads from racing with
schema change by reusing the existing planning lock and planned sink
metadata.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
Validation performed:
- Standard focused FE UT reactor passed `StreamLoadHandlerTest` (14) and
`HttpStreamLoadTablePropertiesTest` (2): 16 tests, zero failures/errors.
Maven Checkstyle passed.
- Unit coverage includes real concurrent base-index replacement with one
read-lock acquisition, metadata read-lock ownership and exceptional
release, the existing planning timeout, first-table response semantics,
and planning without a response. HTTP tests verify planned metadata is
used even when current base-index metadata is absent, for both regular
loads and Group Commit, without another read-lock acquisition.
- Standard ASAN FE/BE build, deployment, and health checks passed in the
personal shared-nothing single-BE environment. P0
`load_p0/http_stream/test_stream_load_schema_metadata` and
`load_p0/http_stream/test_http_stream_properties` passed with zero
failed suites/fatal scripts. Existing regression snapshots were compared
without regeneration or manual edits.
- P0 `schema_change_p0/test_schema_change_unique_mow` also passed: one
suite, zero failed suites/fatal scripts. All three targeted suites used
the binaries built from this revised implementation. Source changes pass
`git diff --check`.
- The concurrency unit test controls the race window; SQL regression
exercises compatibility and does not guarantee it hits that exact
window. This local shared-nothing single-BE environment is not complete
TeamCity or Cloud coverage.
- Behavior changed:
- [ ] No.
- [x] Yes. Response metadata comes from the existing protected planning
scope or the completed HTTP plan, avoiding an unlocked base-index lookup
and additional lock waiting. Multi-table first-table behavior and
protocol fields are preserved.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../org/apache/doris/load/StreamLoadHandler.java | 12 +-
.../apache/doris/service/FrontendServiceImpl.java | 21 ++-
.../apache/doris/load/StreamLoadHandlerTest.java | 170 +++++++++++++++++++++
.../service/HttpStreamLoadTablePropertiesTest.java | 78 ++++++++++
.../test_stream_load_schema_metadata.out | 13 ++
.../test_stream_load_schema_metadata.groovy | 98 ++++++++++++
6 files changed, 378 insertions(+), 14 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
index 8ab0f9db99d..06fde6758dc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
@@ -248,13 +248,6 @@ public class StreamLoadHandler {
if (tables.isEmpty()) {
throw new MetaNotFoundException("table not found");
}
-
- if (result != null) {
- OlapTable olapTable = tables.get(0);
- result.setDbId(db.getId());
- result.setTableId(olapTable.getId());
- result.setBaseSchemaVersion(olapTable.getBaseSchemaVersion());
- }
}
/**
@@ -300,6 +293,11 @@ public class StreamLoadHandler {
result.query_options.setFeProcessUuid(ExecuteEnv.getInstance().getProcessUUID());
result.setIsMowTable(table.getEnableUniqueKeyMergeOnWrite());
result.setEnableTso(table.enableTso());
+ if (this.result != null && fragmentParams.isEmpty()) {
+ this.result.setDbId(db.getId());
+ this.result.setTableId(table.getId());
+ this.result.setBaseSchemaVersion(table.getBaseSchemaVersion());
+ }
fragmentParams.add(result);
if (StringUtils.isEmpty(streamLoadTask.getGroupCommit())) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index de1dd61bb23..03d5a2c9f8d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -270,6 +270,7 @@ import org.apache.doris.thrift.TNodeInfo;
import org.apache.doris.thrift.TNullableStringLiteral;
import org.apache.doris.thrift.TOlapTableIndexTablets;
import org.apache.doris.thrift.TOlapTablePartition;
+import org.apache.doris.thrift.TOlapTableSink;
import org.apache.doris.thrift.TPartitionMeta;
import org.apache.doris.thrift.TPipelineFragmentParams;
import org.apache.doris.thrift.TPipelineWorkloadGroup;
@@ -3190,6 +3191,18 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return httpStreamParams;
}
+ private void setHttpStreamLoadTableProperties(HttpStreamParams
httpStreamParams,
+ TStreamLoadPutResult result) {
+ OlapTable table = (OlapTable) httpStreamParams.getTable();
+ TOlapTableSink sink =
httpStreamParams.getParams().getFragment().getOutputSink().getOlapTableSink();
+
result.getPipelineParams().setIsMowTable(table.getEnableUniqueKeyMergeOnWrite());
+ result.getPipelineParams().setEnableTso(table.enableTso());
+ result.setTableId(sink.getTableId());
+ result.setBaseSchemaVersion(sink.getBaseSchemaVersion());
+ result.setGroupCommitIntervalMs(table.getGroupCommitIntervalMs());
+ result.setGroupCommitDataBytes(table.getGroupCommitDataBytes());
+ }
+
private void httpStreamPutImpl(TStreamLoadPutRequest request,
TStreamLoadPutResult result)
throws UserException {
if (LOG.isDebugEnabled()) {
@@ -3248,14 +3261,8 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
result.getPipelineParams().setTableName(httpStreamParams.getTable().getName());
result.getPipelineParams().setTxnConf(new
TTxnParams().setTxnId(httpStreamParams.getTxnId()));
result.getPipelineParams().setImportLabel(httpStreamParams.getLabel());
- result.getPipelineParams()
- .setIsMowTable(((OlapTable)
httpStreamParams.getTable()).getEnableUniqueKeyMergeOnWrite());
- result.getPipelineParams().setEnableTso(((OlapTable)
httpStreamParams.getTable()).enableTso());
result.setDbId(httpStreamParams.getDb().getId());
- result.setTableId(httpStreamParams.getTable().getId());
- result.setBaseSchemaVersion(((OlapTable)
httpStreamParams.getTable()).getBaseSchemaVersion());
- result.setGroupCommitIntervalMs(((OlapTable)
httpStreamParams.getTable()).getGroupCommitIntervalMs());
- result.setGroupCommitDataBytes(((OlapTable)
httpStreamParams.getTable()).getGroupCommitDataBytes());
+ setHttpStreamLoadTableProperties(httpStreamParams, result);
result.setWaitInternalGroupCommitFinish(Config.wait_internal_group_commit_finish);
} catch (UserException e) {
LOG.warn("exec sql error", e);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
index b8231f963cd..6ec4ea387c0 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
@@ -18,14 +18,22 @@
package org.apache.doris.load;
import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MaterializedIndexMeta;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.cloud.catalog.CloudEnv;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.lock.MonitoredReentrantReadWriteLock;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.mysql.privilege.Auth;
+import org.apache.doris.nereids.load.NereidsStreamLoadPlanner;
+import org.apache.doris.nereids.load.NereidsStreamLoadTask;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
@@ -38,6 +46,7 @@ import org.apache.doris.thrift.TOlapTablePartitionParam;
import org.apache.doris.thrift.TOlapTableSink;
import org.apache.doris.thrift.TPipelineFragmentParams;
import org.apache.doris.thrift.TPlanFragment;
+import org.apache.doris.thrift.TQueryOptions;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
import org.apache.doris.thrift.TTabletLocation;
@@ -45,13 +54,174 @@ import org.apache.doris.thrift.TTabletLocation;
import com.google.common.collect.Lists;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
public class StreamLoadHandlerTest {
+ @Test
+ public void testSchemaVersionLockTimeoutDoesNotReadMetadata() throws
Exception {
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getName()).thenReturn("tbl");
+ Mockito.when(table.tryReadLock(1,
TimeUnit.MILLISECONDS)).thenReturn(false);
+ verifySchemaVersionReadFailure(table, UserException.class);
+ Mockito.verify(table).tryReadLock(1, TimeUnit.MILLISECONDS);
+ Mockito.verify(table, Mockito.never()).getBaseSchemaVersion();
+ Mockito.verify(table, Mockito.never()).readUnlock();
+ }
+
+ @Test
+ public void testSchemaVersionReadFailureReleasesLock() throws Exception {
+ OlapTable table = Mockito.spy(new OlapTable());
+ Mockito.doThrow(new IllegalStateException("schema read
failed")).when(table).getBaseSchemaVersion();
+ verifySchemaVersionReadFailure(table, IllegalStateException.class);
+ MonitoredReentrantReadWriteLock lock = Deencapsulation.getField(table,
"rwLock");
+ Assertions.assertEquals(0, lock.getReadHoldCount());
+ Mockito.verify(table).readUnlock();
+ }
+
+ private void verifySchemaVersionReadFailure(OlapTable table, Class<?
extends Throwable> failure) throws Exception {
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Database database = Mockito.mock(Database.class);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getDbNullable("db")).thenReturn(database);
+ Mockito.when(database.getTableOrMetaException("tbl",
TableType.OLAP)).thenReturn(table);
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setDb("db");
+ request.setTbl("tbl");
+ request.setThriftRpcTimeoutMs(1);
+ StreamLoadHandler handler = new StreamLoadHandler(request, null, new
TStreamLoadPutResult(), "127.0.0.1");
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Deencapsulation.invoke(handler, "setDbAndTable");
+ Assertions.assertThrows(failure, () ->
generateMetadataPlan(handler, table, request));
+ }
+ }
+
+ @Test
+ public void testSchemaVersionReadWaitsForBaseIndexReplacement() throws
Exception {
+ OlapTable table = Mockito.spy(new OlapTable());
+ table.setBaseIndexId(1L);
+ MaterializedIndexMeta oldIndexMeta =
Mockito.mock(MaterializedIndexMeta.class);
+ Mockito.when(oldIndexMeta.getSchemaVersion()).thenReturn(6);
+ table.getMutableIndexIdToMeta().put(1L, oldIndexMeta);
+ MaterializedIndexMeta newIndexMeta =
Mockito.mock(MaterializedIndexMeta.class);
+ Mockito.when(newIndexMeta.getSchemaVersion()).thenReturn(7);
+ table.getMutableIndexIdToMeta().put(2L, newIndexMeta);
+ Mockito.doAnswer(invocation -> {
+ MonitoredReentrantReadWriteLock lock =
Deencapsulation.getField(table, "rwLock");
+ Assertions.assertEquals(1, lock.getReadHoldCount());
+ return invocation.callRealMethod();
+ }).when(table).getBaseSchemaVersion();
+ CountDownLatch oldIndexRemoved = new CountDownLatch(1);
+ CountDownLatch readerStarted = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ readerStarted.countDown();
+ return invocation.callRealMethod();
+ }).when(table).tryReadLock(Mockito.anyLong(),
Mockito.eq(TimeUnit.MILLISECONDS));
+
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Database database = Mockito.mock(Database.class);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ Mockito.when(catalog.getDbNullable("db")).thenReturn(database);
+ Mockito.when(database.getTableOrMetaException("tbl",
TableType.OLAP)).thenReturn(table);
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setDb("db");
+ request.setTbl("tbl");
+ TStreamLoadPutResult result = new TStreamLoadPutResult();
+ StreamLoadHandler handler = new StreamLoadHandler(request, null,
result, "127.0.0.1");
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<?> schemaChange = executor.submit(() -> {
+ table.writeLock();
+ try {
+ // Schema change removes the old base index before switching
to the shadow index.
+ table.getMutableIndexIdToMeta().remove(1L);
+ oldIndexRemoved.countDown();
+ Assertions.assertTrue(readerStarted.await(5,
TimeUnit.SECONDS));
+ table.setBaseIndexId(2L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } finally {
+ table.writeUnlock();
+ }
+ });
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Assertions.assertTrue(oldIndexRemoved.await(5, TimeUnit.SECONDS));
+ Deencapsulation.invoke(handler, "setDbAndTable");
+ // Catalog lookup must not touch index metadata or acquire an
extra read lock.
+ Mockito.verify(table, Mockito.never()).getBaseSchemaVersion();
+ generateMetadataPlan(handler, table, request);
+ Assertions.assertEquals(7, result.getBaseSchemaVersion());
+ Mockito.verify(table, Mockito.times(1)).tryReadLock(5000,
TimeUnit.MILLISECONDS);
+ } finally {
+ readerStarted.countDown();
+ try {
+ schemaChange.get(5, TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+ }
+
+ private void generateMetadataPlan(StreamLoadHandler handler, OlapTable
table,
+ TStreamLoadPutRequest request) throws Exception {
+ NereidsStreamLoadTask task = Mockito.mock(NereidsStreamLoadTask.class);
+ Mockito.when(task.getGroupCommit()).thenReturn("sync_mode");
+ Mockito.doReturn(false).when(table).getEnableUniqueKeyMergeOnWrite();
+ Mockito.doReturn(false).when(table).enableTso();
+ TPipelineFragmentParams plan = buildStreamLoadParams(new
TOlapTableSink());
+ plan.setQueryOptions(new TQueryOptions());
+ try (MockedStatic<NereidsStreamLoadTask> tasks =
Mockito.mockStatic(NereidsStreamLoadTask.class);
+ MockedConstruction<NereidsStreamLoadPlanner> planners =
Mockito.mockConstruction(
+ NereidsStreamLoadPlanner.class, (planner, context) ->
+ Mockito.when(planner.plan(Mockito.any(),
Mockito.anyInt())).thenReturn(plan))) {
+ tasks.when(() ->
NereidsStreamLoadTask.fromTStreamLoadPutRequest(request)).thenReturn(task);
+ handler.generatePlan(table);
+ }
+ }
+
+ @Test
+ public void testResponseKeepsFirstTableMetadata() throws Exception {
+ Database database = new Database(17, "db");
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ TStreamLoadPutResult result = new TStreamLoadPutResult();
+ StreamLoadHandler handler = new StreamLoadHandler(request, null,
result, "127.0.0.1");
+ Deencapsulation.setField(handler, "db", database);
+ OlapTable first = Mockito.spy(new OlapTable());
+ Mockito.doReturn(41L).when(first).getId();
+ Mockito.doReturn(7).when(first).getBaseSchemaVersion();
+ generateMetadataPlan(handler, first, request);
+ OlapTable second = Mockito.spy(new OlapTable());
+ Mockito.doReturn(42L).when(second).getId();
+ generateMetadataPlan(handler, second, request);
+ Assertions.assertEquals(17, result.getDbId());
+ Assertions.assertEquals(41, result.getTableId());
+ Assertions.assertEquals(7, result.getBaseSchemaVersion());
+ Mockito.verify(second, Mockito.never()).getBaseSchemaVersion();
+ }
+
+ @Test
+ public void testPlanningWithoutResponseDoesNotReadExtraSchemaMetadata()
throws Exception {
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ StreamLoadHandler handler = new StreamLoadHandler(request, null, null,
"127.0.0.1");
+ OlapTable table = Mockito.spy(new OlapTable());
+ generateMetadataPlan(handler, table, request);
+ Mockito.verify(table, Mockito.never()).getBaseSchemaVersion();
+ Mockito.verify(table, Mockito.times(1)).tryReadLock(5000,
TimeUnit.MILLISECONDS);
+ }
+
@Test
public void testSelectBackendSkipsDecommissioningBackend() throws
Exception {
SystemInfoService originalSystemInfoService =
Env.getCurrentSystemInfo();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/HttpStreamLoadTablePropertiesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/HttpStreamLoadTablePropertiesTest.java
new file mode 100644
index 00000000000..1ddf9eefac9
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/HttpStreamLoadTablePropertiesTest.java
@@ -0,0 +1,78 @@
+// 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.service;
+
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.qe.HttpStreamParams;
+import org.apache.doris.thrift.TDataSink;
+import org.apache.doris.thrift.TOlapTableSink;
+import org.apache.doris.thrift.TPipelineFragmentParams;
+import org.apache.doris.thrift.TPlanFragment;
+import org.apache.doris.thrift.TStreamLoadPutResult;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.concurrent.TimeUnit;
+
+public class HttpStreamLoadTablePropertiesTest {
+ @Test
+ public void testHttpLoadUsesPlannedSchemaVersion() {
+ verifyPlannedMetadata(23);
+ }
+
+ @Test
+ public void testGroupCommitUsesPlannedSchemaVersion() {
+ verifyPlannedMetadata(-1);
+ }
+
+ private void verifyPlannedMetadata(long txnId) {
+ OlapTable table = Mockito.spy(new OlapTable());
+ // The current table has no base-index metadata. The response must use
the
+ // completed plan, even if schema change has since replaced the index.
+ table.setBaseIndexId(9L);
+ Mockito.doReturn(false).when(table).getEnableUniqueKeyMergeOnWrite();
+ Mockito.doReturn(false).when(table).enableTso();
+ Mockito.doReturn(1000).when(table).getGroupCommitIntervalMs();
+ Mockito.doReturn(1024).when(table).getGroupCommitDataBytes();
+ TOlapTableSink sink = new TOlapTableSink();
+ sink.setTableId(41);
+ sink.setBaseSchemaVersion(7);
+ TPipelineFragmentParams plan = new TPipelineFragmentParams();
+ plan.setFragment(new TPlanFragment().setOutputSink(new
TDataSink().setOlapTableSink(sink)));
+ HttpStreamParams params = new HttpStreamParams();
+ params.setTable(table);
+ params.setTxnId(txnId);
+ params.setParams(plan);
+ TStreamLoadPutResult result = new TStreamLoadPutResult();
+ result.setPipelineParams(plan);
+ FrontendServiceImpl service = Mockito.mock(FrontendServiceImpl.class);
+
+ Deencapsulation.invoke(service, "setHttpStreamLoadTableProperties",
params, result);
+
+ Assertions.assertEquals(41, result.getTableId());
+ Assertions.assertEquals(7, result.getBaseSchemaVersion());
+ Assertions.assertEquals(1000, result.getGroupCommitIntervalMs());
+ Assertions.assertEquals(1024, result.getGroupCommitDataBytes());
+ Mockito.verify(table, Mockito.never()).getBaseSchemaVersion();
+ Mockito.verify(table, Mockito.never()).tryReadLock(Mockito.anyLong(),
Mockito.any(TimeUnit.class));
+ Mockito.verify(table, Mockito.never()).readUnlock();
+ }
+}
diff --git
a/regression-test/data/load_p0/http_stream/test_stream_load_schema_metadata.out
b/regression-test/data/load_p0/http_stream/test_stream_load_schema_metadata.out
new file mode 100644
index 00000000000..f32e8436544
--- /dev/null
+++
b/regression-test/data/load_p0/http_stream/test_stream_load_schema_metadata.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before --
+1 10
+2 20
+
+-- !after_replacement --
+1 10
+2 20
+
+-- !after_addition --
+1 10 7
+2 20 7
+
diff --git
a/regression-test/suites/load_p0/http_stream/test_stream_load_schema_metadata.groovy
b/regression-test/suites/load_p0/http_stream/test_stream_load_schema_metadata.groovy
new file mode 100644
index 00000000000..853abb28461
--- /dev/null
+++
b/regression-test/suites/load_p0/http_stream/test_stream_load_schema_metadata.groovy
@@ -0,0 +1,98 @@
+// 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.
+
+suite("test_stream_load_schema_metadata", "p0") {
+ sql "DROP TABLE IF EXISTS test_stream_load_schema_metadata"
+ sql """
+ CREATE TABLE test_stream_load_schema_metadata (
+ id INT NOT NULL,
+ value INT NOT NULL
+ ) UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "false"
+ )
+ """
+
+ def db = context.config.getDbNameByFile(context.file)
+ def loadRows = { boolean http, boolean groupCommit ->
+ streamLoad {
+ if (http) {
+ set 'version', '1'
+ set 'sql', """
+ INSERT INTO ${db}.test_stream_load_schema_metadata (id,
value)
+ SELECT c1, c2 FROM http_stream("format"="csv",
"column_separator"=",")
+ """
+ } else {
+ table 'test_stream_load_schema_metadata'
+ set 'column_separator', ','
+ set 'columns', 'id,value'
+ }
+ if (groupCommit) {
+ set 'group_commit', 'sync_mode'
+ unset 'label'
+ }
+ inputStream new
ByteArrayInputStream("1,10\n2,20\n".getBytes("UTF-8"))
+ time 10000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(2, json.NumberLoadedRows)
+ assertEquals(0, json.NumberFilteredRows)
+ if (groupCommit) {
+ assertTrue(json.GroupCommit)
+ }
+ }
+ }
+ }
+
+ loadRows(false, false)
+ loadRows(true, false)
+ order_qt_before "SELECT * FROM test_stream_load_schema_metadata"
+
+ // Repeated loads exercise both metadata response paths while the base
index is replaced.
+ sql "ALTER TABLE test_stream_load_schema_metadata MODIFY COLUMN value
BIGINT NOT NULL"
+ for (int i = 0; i < 5; i++) {
+ loadRows(false, false)
+ loadRows(true, false)
+ }
+ waitForSchemaChangeDone {
+ sql """SHOW ALTER TABLE COLUMN WHERE
TableName='test_stream_load_schema_metadata'
+ ORDER BY CreateTime DESC LIMIT 1"""
+ time 120
+ }
+ loadRows(false, false)
+ loadRows(true, false)
+ order_qt_after_replacement "SELECT * FROM test_stream_load_schema_metadata"
+
+ sql "ALTER TABLE test_stream_load_schema_metadata ADD COLUMN added INT
DEFAULT '7'"
+ waitForSchemaChangeDone {
+ sql """SHOW ALTER TABLE COLUMN WHERE
TableName='test_stream_load_schema_metadata'
+ ORDER BY CreateTime DESC LIMIT 1"""
+ time 120
+ }
+ loadRows(false, false)
+ loadRows(true, false)
+ loadRows(true, true)
+ sql "sync"
+ order_qt_after_addition "SELECT * FROM test_stream_load_schema_metadata"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
