This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 18b118249ca branch-4.2: [fix](fe) Refresh S3 streaming source file
metadata #69363 (#69521)
18b118249ca is described below
commit 18b118249cad31506f39e22ff2da2e02bc2017fe
Author: wudi <[email protected]>
AuthorDate: Sat Oct 24 10:30:05 2026 +0800
branch-4.2: [fix](fe) Refresh S3 streaming source file metadata #69363
(#69521)
Cherry-picked from #69363
---
be/src/exec/scan/file_scanner.cpp | 6 +++
be/src/exec/scan/file_scanner_v2.cpp | 7 ++++
.../job/offset/s3/S3EventSourceOffsetProvider.java | 6 +--
.../job/offset/s3/S3SourceOffsetProvider.java | 3 +-
.../ExternalFileTableValuedFunction.java | 26 +++++++++---
.../offset/s3/S3EventSourceOffsetProviderTest.java | 28 +++++++++----
.../job/offset/s3/S3SourceOffsetProviderTest.java | 29 +++++++++++++
.../tablefunction/S3TableValuedFunctionTest.java | 48 +++++++++++++++++++++-
.../test_streaming_insert_job_s3_event_aksk.groovy | 21 +++-------
9 files changed, 139 insertions(+), 35 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index ee5203f16f0..0a172da4216 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -540,6 +540,8 @@ Status FileScanner::_get_block_wrapped(RuntimeState* state,
Block* block, bool*
config::ignore_not_found_file_in_external_table)) {
_cur_reader_eof = true;
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
continue;
} else if (st.is<ErrorCode::END_OF_FILE>()) {
_cur_reader_eof = true;
@@ -580,6 +582,8 @@ Status FileScanner::_get_block_wrapped(RuntimeState* state,
Block* block, bool*
if (st.is<ErrorCode::NOT_FOUND>() &&
config::ignore_not_found_file_in_external_table) {
_cur_reader_eof = true;
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
continue;
}
RETURN_IF_ERROR(st);
@@ -1324,6 +1328,8 @@ Status FileScanner::_get_next_reader() {
if (can_ignore_not_found_file(_current_range,
config::ignore_not_found_file_in_external_table)) {
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
continue;
}
return Status::InternalError("failed to find reader, err: {}",
init_status.to_string());
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index ab055f61dc3..2ad9897a6bb 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -30,6 +30,7 @@
#include "common/cast_set.h"
#include "common/config.h"
#include "common/consts.h"
+#include "common/logging.h"
#include "common/metrics/doris_metrics.h"
#include "common/status.h"
#include "core/assert_cast.h"
@@ -569,6 +570,8 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
_current_range,
config::ignore_not_found_file_in_external_table))) {
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
RETURN_IF_ERROR(_complete_current_split());
_has_prepared_split = false;
block->clear_column_data(cast_set<int64_t>(_projected_columns.size()));
@@ -659,6 +662,8 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
config::ignore_not_found_file_in_external_table))) {
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
RETURN_IF_ERROR(_complete_current_split());
continue;
}
@@ -692,6 +697,8 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
if (ignored_split_status == IgnoredSplitStatus::NOT_FOUND) {
RETURN_IF_ERROR(_table_reader->abort_split());
COUNTER_UPDATE(_not_found_file_counter, 1);
+ LOG(INFO) << "Skip missing file: query_id=" <<
print_id(_state->query_id())
+ << ", path=" << _current_range.path << ",
reason=NOT_FOUND";
RETURN_IF_ERROR(_complete_current_split());
continue;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
index 1a949a9eb59..687127e2f03 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
@@ -572,10 +572,10 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
if (isCommitted(key, message.messageId(),
message.receiptHandle())) {
continue;
}
+ String path = "s3://" + bucket + "/" + key;
files.add(key);
- // The object must remain unchanged until ingestion completes;
notifications identify keys, not versions.
- fileStatuses.add(new TBrokerFileStatus("s3://" + bucket + "/" +
key, false,
- record.getS3().getObject().getSizeAsLong(), true));
+ fileStatuses.add(new TBrokerFileStatus(path, false,
record.getS3().getObject().getSizeAsLong(), true)
+ .setModificationTime(System.currentTimeMillis()));
if (record.getEventTime() != null) {
eventTimestampSeconds = Math.max(eventTimestampSeconds,
record.getEventTime().getMillis() / 1000);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
index 02b22a95ed5..72427f0dd0a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
@@ -97,7 +97,8 @@ public class S3SourceOffsetProvider implements
SourceOffsetProvider {
String beginFile = rfiles.get(0).getName().replace(bucketBase,
"");
String lastFile = rfiles.get(rfiles.size() -
1).getName().replace(bucketBase, "");
offset.setFileStatuses(rfiles.stream().map(file -> new
TBrokerFileStatus(
- file.getName(), file.isDirectory(), file.getSize(),
!file.isDirectory()))
+ file.getName(), file.isDirectory(), file.getSize(),
!file.isDirectory())
+ .setModificationTime(file.getModificationTime()))
.collect(Collectors.toList()));
offset.setStartFile(beginFile);
offset.setEndFile(lastFile);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
index c26bc2f472f..bfdf1d21437 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
@@ -104,6 +104,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -333,7 +334,6 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
return columns;
}
// get one BE address
- columns = Lists.newArrayList();
Backend be = getBackend();
if (be == null) {
String computeGroupHints = "";
@@ -359,17 +359,28 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
TNetworkAddress address = new TNetworkAddress(be.getHost(),
be.getBrpcPort());
try {
- PFetchTableSchemaRequest request = getFetchTableStructureRequest();
+ if (getTFileType() == TFileType.FILE_STREAM) {
+ fileStatuses.add(new TBrokerFileStatus("", false, -1, true));
+ }
+ Iterator<TBrokerFileStatus> files = fileStatuses.iterator();
+ PFetchTableSchemaRequest request =
getFetchTableStructureRequest(files);
InternalService.PFetchTableSchemaResult result = null;
// `request == null` means we don't need to get schemas from BE,
// and we fill a dummy col for this table.
- if (request != null) {
+ while (request != null) {
Future<InternalService.PFetchTableSchemaResult> future =
BackendServiceProxy.getInstance()
.fetchTableStructureAsync(address, request);
result = future.get();
TStatusCode code =
TStatusCode.findByValue(result.getStatus().getStatusCode());
+ // Lance uses a dataset URI rather than an iterator over
individual files.
+ if (code == TStatusCode.NOT_FOUND && !isLanceFormat()) {
+ request = getFetchTableStructureRequest(files);
+ if (request != null) {
+ continue;
+ }
+ }
String errMsg;
if (code != TStatusCode.OK) {
if (!result.getStatus().getErrorMsgsList().isEmpty()) {
@@ -381,7 +392,9 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
}
throw new AnalysisException(errMsg);
}
+ break;
}
+ columns = Lists.newArrayList();
fillColumns(result);
} catch (RpcException e) {
throw new AnalysisException("fetchTableStructureResult rpc
exception", e);
@@ -568,7 +581,8 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
}
}
- private PFetchTableSchemaRequest getFetchTableStructureRequest() throws
TException {
+ private PFetchTableSchemaRequest
getFetchTableStructureRequest(Iterator<TBrokerFileStatus> files)
+ throws TException {
// set TFileScanRangeParams
TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams();
fileScanRangeParams.setFormatType(fileFormatProperties.getFileFormatType());
@@ -597,7 +611,6 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
}
if (getTFileType() == TFileType.FILE_STREAM) {
- fileStatuses.add(new TBrokerFileStatus("", false, -1, true));
fileScanRangeParams.setFileType(getTFileType());
}
@@ -616,7 +629,8 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
firstFile = new TBrokerFileStatus(getFilePath(), false, -1, false);
} else {
firstFile = null;
- for (TBrokerFileStatus fileStatus : fileStatuses) {
+ while (files.hasNext()) {
+ TBrokerFileStatus fileStatus = files.next();
if (isFileContentEmpty(fileStatus)) {
continue;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
index 1bca291f532..d59402b710e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
@@ -271,17 +271,21 @@ public class S3EventSourceOffsetProviderTest {
}
@Test
- public void testNotificationMetadataDoesNotRequireHead() {
+ public void testNotificationUsesEventSizeAndCurrentTimeWithoutHead() {
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
.thenReturn(response(message("literal", record("bucket",
"logs/a%2Cb%5B1%5D.csv", 123))));
+ long before = System.currentTimeMillis();
provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ long after = System.currentTimeMillis();
S3EventOffset offset = nextOffset();
Assertions.assertEquals(Collections.singletonList("logs/a,b[1].csv"),
offset.getFiles());
Assertions.assertEquals("s3://bucket/logs/a,b[1].csv",
offset.getFileStatuses().get(0).getPath());
Assertions.assertEquals(123,
offset.getFileStatuses().get(0).getSize());
-
Assertions.assertFalse(offset.toSerializedJson().contains("fileStatuses"));
+ long modificationTime =
offset.getFileStatuses().get(0).getModificationTime();
+ Assertions.assertTrue(modificationTime >= before && modificationTime
<= after);
Mockito.verify(s3Client,
Mockito.never()).headObject(Mockito.any(HeadObjectRequest.class));
+
Assertions.assertFalse(offset.toSerializedJson().contains("fileStatuses"));
Mockito.verify(sqsClient,
Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
}
@@ -639,9 +643,10 @@ public class S3EventSourceOffsetProviderTest {
}
@Test
- public void
testInvalidRecordsDoNotDiscardValidFilesAndDeletionWaitsForCommit() {
+ public void
testSkippedRecordsDoNotDiscardValidFilesAndDeletionWaitsForCommit() {
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
- .thenReturn(response(message("mixed", record("bucket",
"logs/a.csv", 1), "null",
+ .thenReturn(response(message("mixed", record("bucket",
"logs/missing.csv", 1),
+ record("bucket", "logs/a.csv", 1), "null",
record("bucket", "logs/deleted.csv",
1).replace("ObjectCreated:Put", "ObjectRemoved:Delete"),
"{\"eventSource\":\"aws:s3\",\"eventName\":\"ObjectCreated:Put\"}",
record("bucket", "", 1), record("bucket",
"logs/invalid.csv", -1),
@@ -651,10 +656,13 @@ public class S3EventSourceOffsetProviderTest {
provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
S3EventOffset offset = nextOffset();
- Assertions.assertEquals(Collections.singletonList("logs/a.csv"),
offset.getFiles());
+ Assertions.assertEquals(Arrays.asList("logs/missing.csv",
"logs/a.csv"), offset.getFiles());
+ Assertions.assertEquals(2, offset.getFileStatuses().size());
+ Assertions.assertEquals("s3://bucket/logs/missing.csv",
offset.getFileStatuses().get(0).getPath());
+ Assertions.assertEquals("s3://bucket/logs/a.csv",
offset.getFileStatuses().get(1).getPath());
Mockito.verify(sqsClient,
Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
provider.updateOffset(offset);
- provider.onTaskCommitted(1, 1);
+ provider.onTaskCommitted(2, 2);
Mockito.verify(sqsClient,
Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
ArgumentCaptor<DeleteMessageBatchRequest> deleted =
ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
@@ -783,16 +791,20 @@ public class S3EventSourceOffsetProviderTest {
private void assertResumeReceivesFreshReceipt(boolean running) {
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
- .thenReturn(response(filesMessage("old", "logs/a.csv")),
- response(filesMessage("fresh", "logs/a.csv")));
+ .thenReturn(response(message("old", record("bucket",
"logs/a.csv", 19))),
+ response(message("fresh", record("bucket",
"logs/a.csv", 170))));
provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
if (running) {
nextOffset();
}
provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
Assertions.assertFalse(provider.hasMoreDataToConsume());
+ long before = System.currentTimeMillis();
provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
Assertions.assertEquals(Collections.singletonList("logs/a.csv"),
nextOffset().getFiles());
+ Assertions.assertEquals(170,
nextOffset().getFileStatuses().get(0).getSize());
+ long modificationTime =
nextOffset().getFileStatuses().get(0).getModificationTime();
+ Assertions.assertTrue(modificationTime >= before && modificationTime
<= System.currentTimeMillis());
provider.updateOffset(nextOffset());
provider.onTaskCommitted(1, 1);
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
index 897537366a5..fdd51d00c7b 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
@@ -129,6 +129,35 @@ public class S3SourceOffsetProviderTest {
}
}
+ @Test
+ public void testListingRefreshesFileSizeAndModificationTime() throws
Exception {
+ RemoteFileSystem fs = Mockito.mock(RemoteFileSystem.class);
+ try (MockedStatic<FileSystemFactory> factory =
Mockito.mockStatic(FileSystemFactory.class)) {
+ factory.when(() ->
FileSystemFactory.get(Mockito.any(StorageProperties.class))).thenReturn(fs);
+ for (S3SourceOffsetProvider provider : ImmutableList.of(
+ new S3SourceOffsetProvider(), new
S3SourceOffsetProvider(ONCE_PROPS))) {
+ Mockito.when(fs.globListWithLimit(Mockito.anyString(),
Mockito.anyList(), Mockito.isNull(),
+ Mockito.anyLong(),
Mockito.anyLong())).thenAnswer(invocation -> {
+ List<RemoteFile> files = invocation.getArgument(1);
+ files.add(new RemoteFile("s3://bucket/data/a.csv",
true, 19, 0, 1767225600000L));
+ return new GlobListResult(Status.OK, "data/a.csv",
"bucket", "data/");
+ }).thenAnswer(invocation -> {
+ List<RemoteFile> files = invocation.getArgument(1);
+ files.add(new RemoteFile("s3://bucket/data/a.csv",
true, 170, 0, 1767225601000L));
+ return new GlobListResult(Status.OK, "data/a.csv",
"bucket", "data/");
+ });
+
+ TBrokerFileStatus first = provider.getNextOffset(ONCE_PROPS,
TVF_PROPS).getFileStatuses().get(0);
+ Assertions.assertEquals(19, first.getSize());
+ Assertions.assertEquals(1767225600000L,
first.getModificationTime());
+ // An uncommitted batch must pick up an overwrite on the next
attempt.
+ TBrokerFileStatus retried = provider.getNextOffset(ONCE_PROPS,
TVF_PROPS).getFileStatuses().get(0);
+ Assertions.assertEquals(170, retried.getSize());
+ Assertions.assertEquals(1767225601000L,
retried.getModificationTime());
+ }
+ }
+ }
+
@Test
public void testEmptyAndListingErrorRemainDistinct() throws Exception {
RemoteFileSystem fs = Mockito.mock(RemoteFileSystem.class);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
index 2d79b5a776a..cb388f2e198 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.tablefunction;
+import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.util.S3Util;
@@ -24,19 +25,28 @@ import org.apache.doris.fs.FileSystemFactory;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.trees.expressions.Properties;
import org.apache.doris.nereids.trees.expressions.functions.table.S3;
+import org.apache.doris.proto.InternalService.PFetchTableSchemaRequest;
import org.apache.doris.proto.InternalService.PFetchTableSchemaResult;
+import org.apache.doris.proto.Types.PScalarType;
import org.apache.doris.proto.Types.PStatus;
+import org.apache.doris.proto.Types.PTypeDesc;
+import org.apache.doris.proto.Types.PTypeNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.rpc.BackendServiceProxy;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TPrimitiveType;
import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TTypeNodeType;
import com.google.common.collect.ImmutableMap;
+import org.apache.thrift.TDeserializer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -85,11 +95,40 @@ public class S3TableValuedFunctionTest {
}
@Test
- public void testSchemaFailureDoesNotSkipFiles() throws Exception {
+ public void testSchemaFailurePreservesFilesAndAllowsRetry() throws
Exception {
assertSchemaFailure(TStatusCode.NOT_FOUND);
assertSchemaFailure(TStatusCode.INTERNAL_ERROR);
}
+ @Test
+ public void testSchemaInferenceUsesNextFileAfterNotFound() throws
Exception {
+ S3TableValuedFunction tvf = tvf();
+ BackendServiceProxy proxy = Mockito.mock(BackendServiceProxy.class);
+ PFetchTableSchemaResult schema = PFetchTableSchemaResult.newBuilder()
+
.setStatus(PStatus.newBuilder().setStatusCode(TStatusCode.OK.getValue()))
+ .setColumnNums(1).addColumnNames("id")
+
.addColumnTypes(PTypeDesc.newBuilder().addTypes(PTypeNode.newBuilder()
+ .setType(TTypeNodeType.SCALAR.getValue())
+
.setScalarType(PScalarType.newBuilder().setType(TPrimitiveType.INT.getValue())))).build();
+ Mockito.when(proxy.fetchTableStructureAsync(Mockito.any(),
Mockito.any()))
+
.thenReturn(CompletableFuture.completedFuture(result(TStatusCode.NOT_FOUND)),
+ CompletableFuture.completedFuture(schema));
+ try (MockedStatic<BackendServiceProxy> service =
Mockito.mockStatic(BackendServiceProxy.class)) {
+ service.when(BackendServiceProxy::getInstance).thenReturn(proxy);
+ Assertions.assertEquals("id",
tvf.getTableColumns().get(0).getName());
+ Assertions.assertEquals(PrimitiveType.INT,
tvf.getTableColumns().get(0).getType().getPrimitiveType());
+ ArgumentCaptor<PFetchTableSchemaRequest> requests =
+ ArgumentCaptor.forClass(PFetchTableSchemaRequest.class);
+ Mockito.verify(proxy,
Mockito.times(2)).fetchTableStructureAsync(Mockito.any(), requests.capture());
+ for (int i = 0; i < 2; i++) {
+ TFileScanRange range = new TFileScanRange();
+ new TDeserializer().deserialize(range,
requests.getAllValues().get(i).getFileScanRange().toByteArray());
+
Assertions.assertEquals(tvf.getFileStatuses().get(i).getPath(),
range.getRanges().get(0).getPath());
+ }
+ Assertions.assertEquals(2, tvf.getFileStatuses().size());
+ }
+ }
+
@Test
public void testExactFilesAreScopedToTvf() throws Exception {
S3TableValuedFunction first = new S3TableValuedFunction(PROPERTIES,
@@ -120,7 +159,12 @@ public class S3TableValuedFunctionTest {
try (MockedStatic<BackendServiceProxy> service =
Mockito.mockStatic(BackendServiceProxy.class)) {
service.when(BackendServiceProxy::getInstance).thenReturn(proxy);
Assertions.assertThrows(AnalysisException.class,
tvf::getTableColumns);
- Mockito.verify(proxy).fetchTableStructureAsync(Mockito.any(),
Mockito.any());
+ int attempts = code == TStatusCode.NOT_FOUND ? 2 : 1;
+ Mockito.verify(proxy, Mockito.times(attempts))
+ .fetchTableStructureAsync(Mockito.any(), Mockito.any());
+ Assertions.assertThrows(AnalysisException.class,
tvf::getTableColumns);
+ Mockito.verify(proxy, Mockito.times(attempts * 2))
+ .fetchTableStructureAsync(Mockito.any(), Mockito.any());
Assertions.assertEquals(2, tvf.getFileStatuses().size());
}
}
diff --git
a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
index 6b519b621a8..b92c4b072f7 100644
---
a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
+++
b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
@@ -102,6 +102,11 @@ suite("test_streaming_insert_job_s3_event_aksk") {
}
try {
sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ // Observe the invalid message before starting a consumer; PAUSE can
leave a receive in flight.
+
sqsClient.sendMessage(SendMessageRequest.builder().queueUrl(queueUrl).messageBody("{}").build())
+ Awaitility.await().pollInSameThread().atMost(120,
SECONDS).pollInterval(1, SECONDS).until {
+
queueCounts()[QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES].toLong() > 0
+ }
sql "DROP TABLE IF EXISTS test_s3_event_aksk"
sql """CREATE TABLE test_s3_event_aksk (id INT, value STRING)
UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
@@ -131,6 +136,7 @@ suite("test_streaming_insert_job_s3_event_aksk") {
)"""
awaitRows(1)
awaitOffsets(["z.csv"])
+ awaitQueueEmpty()
order_qt_first "SELECT * FROM test_s3_event_aksk"
// A smaller key uploaded after the first commit must still be
discovered in NOTIFICATION mode.
@@ -191,21 +197,6 @@ suite("test_streaming_insert_job_s3_event_aksk") {
}
metricsReady
}
-
- String committedOffset = sql("SELECT CurrentOffset FROM
jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")[0][0]
- // A message without S3 records is acknowledged without advancing the
job.
- sql "PAUSE JOB WHERE jobname='${jobName}'"
-
sqsClient.sendMessage(SendMessageRequest.builder().queueUrl(queueUrl).messageBody("{}").build())
- Awaitility.await().pollInSameThread().atMost(120,
SECONDS).pollInterval(1, SECONDS).until {
-
queueCounts()[QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES].toLong() > 0
- }
- sql "RESUME JOB WHERE jobname='${jobName}'"
- awaitQueueEmpty()
- Awaitility.await().pollInSameThread().atMost(120,
SECONDS).pollInterval(1, SECONDS).until {
- def job = sql("SELECT Status, CurrentOffset FROM
jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
- job[0][0] == "RUNNING" && job[0][1] == committedOffset &&
- (sql("SELECT COUNT(*) FROM test_s3_event_aksk")[0][0] as
long) == 5
- }
} finally {
try {
sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
