Skip to content

branch-4.1: [fix](fe) Refresh S3 streaming source file metadata #69363 #69519

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JNSimba wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
+139 −35
Open

branch-4.1: [fix](fe) Refresh S3 streaming source file metadata #69363 #69519

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions be/src/exec/scan/file_scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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());
Expand Down
7 changes: 7 additions & 0 deletions be/src/exec/scan/file_scanner_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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()));
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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;
}
Expand Down
6 changes: 3 additions & 3 deletions fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,8 @@ public S3Offset getNextOffset(StreamingJobProperties jobProps, Map<String, Strin
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);
Expand Down
26 changes: 20 additions & 6 deletions fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -336,7 +337,6 @@ public List<Column> getTableColumns() throws AnalysisException {
return columns;
}
// get one BE address
columns = Lists.newArrayList();
Backend be = getBackend();
if (be == null) {
String computeGroupHints = "";
Expand All @@ -362,17 +362,28 @@ public List<Column> 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<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()) {
Expand All @@ -384,7 +395,9 @@ public List<Column> getTableColumns() throws AnalysisException {
}
throw new AnalysisException(errMsg);
}
break;
}
columns = Lists.newArrayList();
fillColumns(result);
} catch (RpcException e) {
throw new AnalysisException("fetchTableStructureResult rpc exception", e);
Expand Down Expand Up @@ -569,7 +582,8 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) {
}
}

private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TException {
private PFetchTableSchemaRequest getFetchTableStructureRequest(Iterator<TBrokerFileStatus> files)
throws TException {
// set TFileScanRangeParams
TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams();
fileScanRangeParams.setFormatType(fileFormatProperties.getFileFormatType());
Expand Down Expand Up @@ -598,7 +612,6 @@ private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TExcepti
}

if (getTFileType() == TFileType.FILE_STREAM) {
fileStatuses.add(new TBrokerFileStatus("", false, -1, true));
fileScanRangeParams.setFileType(getTFileType());
}

Expand All @@ -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;
}
Expand Down
28 changes: 20 additions & 8 deletions fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}

Expand Down Expand Up @@ -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),
Expand All @@ -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<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
Expand Down Expand Up @@ -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)))
Expand Down
29 changes: 29 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,35 @@ public void testOnceMarksLastBatchAndRecovers() throws Exception {
}
}

@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);
Expand Down
Loading
Loading