This is an automated email from the ASF dual-hosted git repository.
JNSimba 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 18945e21408 [fix](fe) Refresh S3 streaming source file metadata
(#69363)
18945e21408 is described below
commit 18945e21408ac1b653b8df1a567b866248aa11e2
Author: wudi <[email protected]>
AuthorDate: Fri Oct 23 16:46:31 2026 +0800
[fix](fe) Refresh S3 streaming source file metadata (#69363)
### What problem does this PR solve?
Problem Summary:
S3 streaming LIST and ONE_TIME ingestion discard the listed modification
time when building scan metadata. After an object is overwritten and
ingestion resumes, the unchanged cache identity can reuse bytes from the
previous object. NOTIFICATION ingestion also leaves the modification
time unset. Separately, schema inference fails when its first candidate
file has disappeared, even if another file in the batch remains
readable.
This PR:
- Preserves the listed modification time in LIST and ONE_TIME scan
metadata.
- Sets the local wall-clock time as the cache timestamp when parsing an
accepted NOTIFICATION record. The size still comes from the
notification; no HEAD request is added.
- Tries the next schema candidate after a NOT_FOUND response and leaves
the schema cache unset when inference fails, so a later attempt can
retry.
- Adds an INFO log with query ID, path and reason at each existing BE
missing-file skip site. The existing skip policy and
transaction/acknowledgement flow are unchanged.
- Moves the notification regression's invalid-message check before job
creation to avoid racing an in-flight receive during PAUSE.
Known limitations:
- An old notification can still carry a stale size after a same-key
overwrite. Refreshing the cache timestamp does not refresh that byte
bound. If old and replacement notifications are coalesced, same-key
deduplication can retain the old size and acknowledge both events after
a truncated read. This remains unresolved.
- Without an explicit schema, inference still fails if every eligible
candidate returns NOT_FOUND; this PR does not add an empty-batch
acknowledgement path.
- Cache timestamps do not provide object-version isolation or guarantee
distinct identities for every replacement. Objects must remain unchanged
during an active ingestion batch.
---
be/src/exec/scan/file_scanner.cpp | 6 +++
be/src/exec/scan/file_scanner_v2.cpp | 5 +++
.../job/offset/s3/S3EventSourceOffsetProvider.java | 6 +--
.../job/offset/s3/S3SourceOffsetProvider.java | 3 +-
.../ExternalFileTableValuedFunction.java | 25 ++++++++---
.../offset/s3/S3EventSourceOffsetProviderTest.java | 28 +++++++++----
.../job/offset/s3/S3SourceOffsetProviderTest.java | 25 +++++++++++
.../tablefunction/S3TableValuedFunctionTest.java | 48 +++++++++++++++++++++-
.../test_streaming_insert_job_s3_event_aksk.groovy | 21 +++-------
9 files changed, 132 insertions(+), 35 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index 3a4fc99a6ec..15ae816cad2 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -552,6 +552,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;
@@ -588,6 +590,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;
}
RETURN_IF_ERROR(st);
@@ -1305,6 +1309,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 c20e2f92cb5..c8aaab44f7a 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"
@@ -512,6 +513,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";
_state->update_num_finished_scan_range(1);
_has_prepared_split = false;
block->clear_column_data(cast_set<int64_t>(_projected_columns.size()));
@@ -601,6 +604,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";
_state->update_num_finished_scan_range(1);
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 c4b6f20d5a6..b744a248007 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
@@ -573,10 +573,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.s3.object.size, true));
+ fileStatuses.add(new TBrokerFileStatus(path, false,
record.s3.object.size, true)
+ .setModificationTime(System.currentTimeMillis()));
if (record.eventTime != null) {
eventTimestampSeconds = Math.max(eventTimestampSeconds,
record.eventTime.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 dcc84a5931b..1a4149ead45 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
@@ -91,7 +91,8 @@ public class S3SourceOffsetProvider implements
SourceOffsetProvider {
String beginFile =
rfiles.get(0).location().uri().replace(bucketBase, "");
String lastFile = rfiles.get(rfiles.size() -
1).location().uri().replace(bucketBase, "");
offset.setFileStatuses(rfiles.stream().map(entry -> new
TBrokerFileStatus(
- entry.location().uri(), entry.isDirectory(),
entry.length(), !entry.isDirectory()))
+ entry.location().uri(), entry.isDirectory(),
entry.length(), !entry.isDirectory())
+ .setModificationTime(entry.modificationTime()))
.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 3f15a190ccd..81d7f834c78 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
@@ -97,6 +97,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -310,7 +311,6 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
return columns;
}
// get one BE address
- columns = Lists.newArrayList();
Backend be = getBackend();
if (be == null) {
String computeGroupHints = "";
@@ -336,17 +336,27 @@ 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());
+ if (code == TStatusCode.NOT_FOUND) {
+ request = getFetchTableStructureRequest(files);
+ if (request != null) {
+ continue;
+ }
+ }
String errMsg;
if (code != TStatusCode.OK) {
if (!result.getStatus().getErrorMsgsList().isEmpty()) {
@@ -358,7 +368,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);
@@ -506,7 +518,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());
@@ -527,7 +540,6 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
}
if (getTFileType() == TFileType.FILE_STREAM) {
- fileStatuses.add(new TBrokerFileStatus("", false, -1, true));
fileScanRangeParams.setFileType(getTFileType());
}
@@ -541,7 +553,8 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
// get first file, used to parse table schema
TBrokerFileStatus 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 0c095d0678f..a9e4f667698 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));
}
@@ -685,9 +689,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),
@@ -697,10 +702,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);
@@ -829,16 +837,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 48f479ae92f..863c144c1da 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
@@ -125,6 +125,31 @@ public class S3SourceOffsetProviderTest {
}
}
+ @Test
+ public void testListingRefreshesFileSizeAndModificationTime() throws
Exception {
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ try (MockedStatic<FileSystemFactory> factory =
Mockito.mockStatic(FileSystemFactory.class)) {
+ factory.when(() ->
FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs);
+ for (S3SourceOffsetProvider provider : List.of(
+ new S3SourceOffsetProvider(), new
S3SourceOffsetProvider(ONCE_PROPS))) {
+ Mockito.when(fs.globListWithLimit(Mockito.any(Location.class),
Mockito.isNull(),
+ Mockito.anyLong(), Mockito.anyLong())).thenReturn(
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 19, false, 1767225600000L, null)),
"bucket", "data/", "data/a.csv"),
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 170, false, 1767225601000L, null)),
"bucket", "data/", "data/a.csv"));
+
+ 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 {
FileSystem fs = Mockito.mock(FileSystem.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 fb46e779dd8..82313b4c126 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,18 +25,27 @@ 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 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;
@@ -84,11 +94,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,
@@ -119,7 +158,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]
