diff --git a/be/src/exec/scan/file_scanner.cpp b/be/src/exec/scan/file_scanner.cpp index ee5203f16f0930..0a172da4216b85 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()) { _cur_reader_eof = true; @@ -580,6 +582,8 @@ Status FileScanner::_get_block_wrapped(RuntimeState* state, Block* block, bool* if (st.is() && 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 ab055f61dc3d93..2ad9897a6bbc2a 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(_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 1a949a9eb59870..687127e2f0394f 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 @@ private ParsedMessage parseMessage(Message message) { 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 02b22a95ed592b..72427f0dd0a4ef 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 S3Offset getNextOffset(StreamingJobProperties jobProps, Map 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 41466e0648a718..a435f62f05b407 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.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; @@ -336,7 +337,6 @@ public List getTableColumns() throws AnalysisException { return columns; } // get one BE address - columns = Lists.newArrayList(); Backend be = getBackend(); if (be == null) { String computeGroupHints = ""; @@ -362,17 +362,28 @@ public List getTableColumns() throws AnalysisException { 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 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 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()) { @@ -384,7 +395,9 @@ public List getTableColumns() throws AnalysisException { } throw new AnalysisException(errMsg); } + break; } + columns = Lists.newArrayList(); fillColumns(result); } catch (RpcException e) { throw new AnalysisException("fetchTableStructureResult rpc exception", e); @@ -569,7 +582,8 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) { } } - private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TException { + private PFetchTableSchemaRequest getFetchTableStructureRequest(Iterator files) + throws TException { // set TFileScanRangeParams TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams(); fileScanRangeParams.setFormatType(fileFormatProperties.getFileFormatType()); @@ -598,7 +612,6 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TExcepti } if (getTFileType() == TFileType.FILE_STREAM) { - fileStatuses.add(new TBrokerFileStatus("", false, -1, true)); fileScanRangeParams.setFileType(getTFileType()); } @@ -617,7 +630,8 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TExcepti 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 1bca291f532719..d59402b710e973 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 void testSqsCloseFailureStillClosesFileSystem() { } @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 void testInvalidMessagesAreDeletedWithoutPublishingBatch() { } @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 void testInvalidRecordsDoNotDiscardValidFilesAndDeletionWaitsForCommit() 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 deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class); @@ -783,16 +791,20 @@ public void testResumeDiscardsRunningBatchAndReceivesFreshReceipt() { 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 897537366a5839..fdd51d00c7b00b 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 void testOnceMarksLastBatchAndRecovers() throws Exception { } } + @Test + public void testListingRefreshesFileSizeAndModificationTime() throws Exception { + RemoteFileSystem fs = Mockito.mock(RemoteFileSystem.class); + try (MockedStatic 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 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 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 2d79b5a776a0de..cb388f2e198f60 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.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 void testExactFilesKeepLiteralPathsAndSkipListing() throws Exception { } @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 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 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 @@ private void assertSchemaFailure(TStatusCode code) throws Exception { try (MockedStatic 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 6b519b621a84a4..b92c4b072f7841 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}'"