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
commit a946eefa6c2dcddca13a444337f1124dcd1845e8 Author: wudi <[email protected]> AuthorDate: Tue Oct 13 10:21:53 2026 +0800 branch-4.1: [improvement](streaming) Support one-time S3 streaming ingestion #68007 (#68943) Cherry-picked from #68007 --- .../insert/streaming/StreamingInsertJob.java | 76 +++++++---- .../insert/streaming/StreamingJobProperties.java | 25 +++- .../streaming/StreamingJobSchedulerTask.java | 4 +- .../job/offset/SourceOffsetProviderFactory.java | 7 +- .../org/apache/doris/job/offset/s3/S3Offset.java | 2 + .../job/offset/s3/S3SourceOffsetProvider.java | 30 +++- .../job/scheduler/StreamingTaskScheduler.java | 6 +- .../trees/plans/commands/AlterJobCommand.java | 19 ++- .../StreamingInsertJobOffsetPersistenceTest.java | 105 +++++++++++++- .../StreamingInsertJobStatusTransitionTest.java | 36 +++++ .../streaming/StreamingJobPropertiesTest.java | 43 ++++++ .../job/offset/s3/S3SourceOffsetProviderTest.java | 151 +++++++++++++++++++++ .../job/scheduler/StreamingTaskSchedulerTest.java | 70 ++++++++++ .../test_streaming_insert_job_s3_once.groovy | 107 +++++++++++++++ 14 files changed, 642 insertions(+), 39 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java index 13db794b4ca..972be838e19 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java @@ -404,7 +404,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M private SourceOffsetProvider createOffsetProvider(Map<String, String> jdbcSourceProps) { SourceOffsetProvider provider; if (tvfType != null) { - provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType); + provider = SourceOffsetProviderFactory.createSourceOffsetProvider(tvfType, jobProperties); } else { provider = new JdbcSourceOffsetProvider(getJobId(), dataSourceType, jdbcSourceProps); } @@ -555,6 +555,9 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M public void updateJobStatus(JobStatus status) throws JobException { lock.writeLock().lock(); try { + if (isFinalStatus() && !getJobStatus().equals(status)) { + throw new JobException("Can't update final job status " + getJobStatus() + " to " + status); + } super.updateJobStatus(status); if (JobStatus.PAUSED.equals(getJobStatus())) { clearRunningStreamTask(status); @@ -568,6 +571,41 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M } } + public boolean tryFinishJob() throws JobException { + lock.writeLock().lock(); + try { + if (!isActive()) { + return false; + } + if (runningStreamTask != null && TaskStatus.PENDING.equals(runningStreamTask.getStatus())) { + // Cancel the waiting task when a metadata scan detects the end of the source. + cancelAllTasks(false); + } + resetFailureInfo(null); + updateJobStatus(JobStatus.FINISHED); + logUpdateOperation(); + return true; + } finally { + lock.writeLock().unlock(); + } + } + + private boolean tryPauseJob(FailureReason reason) throws JobException { + lock.writeLock().lock(); + try { + if (!isActive() + || (getFailureReason() != null + && InternalErrorCode.MANUAL_PAUSE_ERR.equals(getFailureReason().getCode()))) { + return false; + } + updateJobStatus(JobStatus.PAUSED); + setFailureReason(reason); + return true; + } finally { + lock.writeLock().unlock(); + } + } + public void resetFailureInfo(FailureReason reason) { this.setFailureReason(reason); // Currently, only delayMsg is present here, which needs to be cleared when the status changes. @@ -742,17 +780,8 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M offsetProvider.fetchRemoteMeta(props); } catch (Exception ex) { log.warn("fetch remote meta failed, job id: {}", getJobId(), ex); - if (this.getFailureReason() == null - || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) { - // When a job is manually paused, it does not need to be set again, - // otherwise, it may be woken up by auto resume. - // Pause before setting the reason: updateJobStatus's writeLock orders this after any - // task-success callback that clears failureReason, so a success can't wipe the reason. - this.updateJobStatus(JobStatus.PAUSED); - this.setFailureReason( - new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, - "Failed to fetch meta, " + ex.getMessage())); - + if (tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, + "Failed to fetch meta, " + ex.getMessage()))) { if (MetricRepo.isInit) { MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L); } @@ -790,26 +819,24 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M } } catch (Exception ex) { log.warn("advance splits failed, job id: {}", getJobId(), ex); - if (this.getFailureReason() == null - || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) { - this.setFailureReason(new FailureReason( - InternalErrorCode.GET_REMOTE_DATA_ERROR, - "Failed to advance splits, " + ex.getMessage())); - this.updateJobStatus(JobStatus.PAUSED); - } + tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, + "Failed to advance splits, " + ex.getMessage())); } } public boolean needScheduleTask() { readLock(); try { - return (getJobStatus().equals(JobStatus.RUNNING) - || getJobStatus().equals(JobStatus.PENDING)); + return isActive(); } finally { readUnlock(); } } + private boolean isActive() { + return JobStatus.PENDING.equals(getJobStatus()) || JobStatus.RUNNING.equals(getJobStatus()); + } + public void clearRunningStreamTask(JobStatus newJobStatus) { if (runningStreamTask != null) { log.info("clear running streaming insert task for job {}, task {}, status {} ", @@ -948,6 +975,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M this.jobStatistic.setFileSize(attachment.getFileBytes()); this.jobStatistic.setFilteredRows(attachment.getFilteredRows()); offsetProvider.updateOffset(offsetProvider.deserializeOffset(attachment.getOffset())); + this.offsetProviderPersist = offsetProvider.getPersistInfo(); //update metric if (MetricRepo.isInit && !isReplay) { @@ -1458,6 +1486,9 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M @Override public void gsonPostProcess() throws IOException { + if (jobProperties == null && properties != null) { + jobProperties = new StreamingJobProperties(properties); + } if (offsetProvider == null) { offsetProvider = createOffsetProvider(sourceProperties); if (tvfType != null) { @@ -1465,9 +1496,6 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M } } - if (jobProperties == null && properties != null) { - jobProperties = new StreamingJobProperties(properties); - } recomputeDerivedFields(); if (null == getSucceedTaskCount()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java index 1bde1bf6205..d0ea8ec6055 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java @@ -33,6 +33,7 @@ import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; @@ -41,12 +42,16 @@ public class StreamingJobProperties implements JobProperties { public static final String MAX_INTERVAL_SECOND_PROPERTY = "max_interval"; public static final String S3_MAX_BATCH_FILES_PROPERTY = "s3.max_batch_files"; public static final String S3_MAX_BATCH_BYTES_PROPERTY = "s3.max_batch_bytes"; + public static final String S3_INGESTION_MODE_PROPERTY = "s3.ingestion_mode"; + public static final String S3_INGESTION_MODE_LEXICAL = "LEXICAL"; + public static final String S3_INGESTION_MODE_ONCE = "ONCE"; public static final String SESSION_VAR_PREFIX = "session."; public static final String INTERNAL_KEY_PREFIX = "__"; public static final String OFFSET_PROPERTY = "offset"; public static final String COMPUTE_GROUP_PROPERTY = "compute_group"; public static final List<String> SUPPORT_STREAM_JOB_PROPS = Arrays.asList(MAX_INTERVAL_SECOND_PROPERTY, - S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY); + S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, S3_INGESTION_MODE_PROPERTY, + OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY); public static final long DEFAULT_MAX_INTERVAL_SECOND = 10; public static final long DEFAULT_MAX_S3_BATCH_FILES = 256; @@ -123,6 +128,15 @@ public class StreamingJobProperties implements JobProperties { + " must be between 100 MB and 10 GB, but was " + properties.get(StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY)); + String ingestionMode = getS3IngestionMode(); + if (!S3_INGESTION_MODE_LEXICAL.equals(ingestionMode) + && !S3_INGESTION_MODE_ONCE.equals(ingestionMode)) { + throw new AnalysisException("Unsupported s3.ingestion_mode: " + ingestionMode); + } + if (S3_INGESTION_MODE_ONCE.equals(ingestionMode) && properties.containsKey(OFFSET_PROPERTY)) { + throw new AnalysisException("offset is not supported when s3.ingestion_mode is ONCE"); + } + // validate session variables try { Map<String, String> sessionVarMap = parseSessionVarMap(); @@ -204,4 +218,13 @@ public class StreamingJobProperties implements JobProperties { public String getComputeGroup() { return properties.get(COMPUTE_GROUP_PROPERTY); } + + public String getS3IngestionMode() { + return properties.getOrDefault(S3_INGESTION_MODE_PROPERTY, S3_INGESTION_MODE_LEXICAL) + .trim().toUpperCase(Locale.ROOT); + } + + public boolean isS3OnceMode() { + return S3_INGESTION_MODE_ONCE.equals(getS3IngestionMode()); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java index a017b0bb921..c33ce7f5895 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java @@ -74,9 +74,7 @@ public class StreamingJobSchedulerTask extends AbstractTask { } if (streamingInsertJob.hasReachedEnd()) { // Source already fully consumed (e.g. snapshot-only mode recovered after FE restart). - // Transition directly to FINISHED without creating a new task. - streamingInsertJob.updateJobStatus(JobStatus.FINISHED); - streamingInsertJob.logUpdateOperation(); + streamingInsertJob.tryFinishJob(); return; } streamingInsertJob.createStreamingTask(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java index 30f9d0edd57..631162d27fe 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java @@ -18,6 +18,7 @@ package org.apache.doris.job.offset; import org.apache.doris.job.exception.JobException; +import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties; import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider; import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; @@ -35,8 +36,12 @@ public class SourceOffsetProviderFactory { map.put("cdc_stream", JdbcTvfSourceOffsetProvider.class); } - public static SourceOffsetProvider createSourceOffsetProvider(String sourceType) { + public static SourceOffsetProvider createSourceOffsetProvider( + String sourceType, StreamingJobProperties jobProperties) { try { + if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3OnceMode()) { + return new S3SourceOffsetProvider(jobProperties); + } Class<? extends SourceOffsetProvider> cla = map.get(sourceType.toLowerCase()); if (cla == null) { throw new JobException("Unsupported source type: " + sourceType); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java index ebd33803490..44df1756e02 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java @@ -34,6 +34,8 @@ public class S3Offset implements Offset { // s3://bucket/path/{1.csv,2.csv} String fileLists; int fileNum; + @SerializedName("lastBatch") + boolean lastBatch; @Override public String toSerializedJson() { 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 a3eed67b378..a89affb9536 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 @@ -50,8 +50,18 @@ import java.util.stream.Collectors; @Log4j2 public class S3SourceOffsetProvider implements SourceOffsetProvider { - S3Offset currentOffset; - String maxEndFile; + private final boolean onceMode; + private volatile S3Offset noMoreFilesAfterOffset; + volatile S3Offset currentOffset; + volatile String maxEndFile; + + public S3SourceOffsetProvider() { + this.onceMode = false; + } + + public S3SourceOffsetProvider(StreamingJobProperties jobProperties) { + this.onceMode = jobProperties.isS3OnceMode(); + } @Override public String getSourceType() { @@ -101,6 +111,7 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider { offset.setEndFile(lastFile); offset.setFileNum(rfiles.size()); maxEndFile = globListResult.getMaxFile(); + offset.setLastBatch(onceMode && lastFile.equals(globListResult.getMaxFile())); } else { throw new RuntimeException("No new files found in path: " + filePath); } @@ -165,7 +176,8 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider { Map<String, String> copiedProps = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); copiedProps.putAll(properties); StorageProperties storageProperties = StorageProperties.createPrimary(copiedProps); - String startFile = currentOffset == null ? null : currentOffset.endFile; + S3Offset offsetAtScan = currentOffset; + String startFile = offsetAtScan == null ? null : offsetAtScan.endFile; try (RemoteFileSystem fileSystem = FileSystemFactory.get(storageProperties)) { String uri = storageProperties.validateAndGetUri(copiedProps); String filePath = storageProperties.validateAndNormalizeUri(uri); @@ -181,6 +193,9 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider { ? globListResult.getStatus().getErrMsg() : "null result"; throw new Exception("Failed to list S3 files: " + errMsg); } + if (onceMode) { + noMoreFilesAfterOffset = objects.isEmpty() ? offsetAtScan : null; + } if (!objects.isEmpty() && StringUtils.isNotEmpty(globListResult.getMaxFile())) { maxEndFile = globListResult.getMaxFile(); } @@ -189,6 +204,9 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider { @Override public boolean hasMoreDataToConsume() { + if (hasReachedEnd()) { + return false; + } if (currentOffset == null || currentOffset.endFile == null) { return true; } @@ -199,6 +217,12 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider { return false; } + @Override + public boolean hasReachedEnd() { + S3Offset offset = currentOffset; + return onceMode && offset != null && (offset.isLastBatch() || noMoreFilesAfterOffset == offset); + } + @Override public String getPersistInfo() { if (currentOffset == null) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java index 91d5fb1a658..62d324061b0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java @@ -104,7 +104,7 @@ public class StreamingTaskScheduler extends MasterDaemon { } } - private void scheduleOneTask(AbstractStreamingTask task) { + private void scheduleOneTask(AbstractStreamingTask task) throws JobException { if (DebugPointUtil.isEnable("StreamingJob.scheduleTask.exception")) { throw new RuntimeException("debug point StreamingJob.scheduleTask.exception"); } @@ -122,6 +122,10 @@ public class StreamingTaskScheduler extends MasterDaemon { } // reject task if no more data to consume if (!job.hasMoreDataToConsume()) { + if (job.hasReachedEnd()) { + job.tryFinishJob(); + return; + } String delayMsg = "No data available for consumption at the moment, will retry after " + (System.currentTimeMillis() + DELAY_SCHEDULER_MS); job.setJobRuntimeMsg(delayMsg); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java index 0a529d55424..e31af87bb36 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java @@ -41,6 +41,7 @@ import org.apache.doris.qe.StmtExecutor; import com.google.common.base.Preconditions; import org.apache.commons.lang3.StringUtils; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -277,11 +278,19 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne } private void validateProps(StreamingInsertJob streamingJob) throws AnalysisException { - StreamingJobProperties jobProperties = new StreamingJobProperties(properties); - jobProperties.validate(); - if (jobProperties.getOffsetProperty() != null) { - streamingJob.validateAlterOffset(jobProperties.getOffsetProperty()); - streamingJob.validateOffset(jobProperties.getOffsetProperty()); + StreamingJobProperties originJobProperties = + new StreamingJobProperties(streamingJob.getProperties()); + Map<String, String> mergedProperties = new HashMap<>(streamingJob.getProperties()); + mergedProperties.putAll(properties); + StreamingJobProperties updatedJobProperties = new StreamingJobProperties(mergedProperties); + updatedJobProperties.validate(); + if (!originJobProperties.getS3IngestionMode().equals(updatedJobProperties.getS3IngestionMode())) { + throw new AnalysisException("s3.ingestion_mode cannot be altered"); + } + String offset = properties.get(StreamingJobProperties.OFFSET_PROPERTY); + if (offset != null) { + streamingJob.validateAlterOffset(offset); + streamingJob.validateOffset(offset); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java index 0b13d0ec745..fde21328560 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java @@ -27,10 +27,15 @@ import org.apache.doris.job.common.TaskStatus; import org.apache.doris.job.exception.JobException; import org.apache.doris.job.manager.JobManager; import org.apache.doris.job.manager.StreamingTaskManager; +import org.apache.doris.job.offset.SourceOffsetProvider; import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider; +import org.apache.doris.job.offset.s3.S3Offset; +import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; +import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.transaction.GlobalTransactionMgrIface; import org.apache.doris.transaction.TxnStateCallbackFactory; +import com.google.common.collect.ImmutableMap; import org.junit.Assert; import org.junit.Test; import org.mockito.MockedStatic; @@ -38,6 +43,7 @@ import org.mockito.Mockito; import java.util.Collections; import java.util.HashMap; +import java.util.Map; import java.util.concurrent.locks.ReentrantReadWriteLock; public class StreamingInsertJobOffsetPersistenceTest { @@ -163,6 +169,103 @@ public class StreamingInsertJobOffsetPersistenceTest { } } + @Test + public void testS3OnceLastBatchFinishesOnlyAfterSuccess() throws Exception { + StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", "ONCE")); + TestStreamingInsertJob failedJob = newJob(new S3SourceOffsetProvider(properties), 1017L); + NoopStreamingMultiTblTask failedTask = + (NoopStreamingMultiTblTask) Deencapsulation.getField(failedJob, "runningStreamTask"); + S3Offset failedOffset = new S3Offset(); + failedOffset.setLastBatch(true); + Deencapsulation.setField(failedTask, "runningOffset", failedOffset); + failedTask.setErrMsg("failed"); + + TestStreamingInsertJob succeededJob = newJob(new S3SourceOffsetProvider(properties), 1018L); + NoopStreamingMultiTblTask succeededTask = + (NoopStreamingMultiTblTask) Deencapsulation.getField(succeededJob, "runningStreamTask"); + S3Offset succeededOffset = new S3Offset(); + succeededOffset.setEndFile("data/b.csv"); + succeededOffset.setLastBatch(true); + Deencapsulation.setField(succeededTask, "runningOffset", succeededOffset); + + try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class)) { + Env env = Mockito.mock(Env.class); + JobManager<?, ?> jobManager = Mockito.mock(JobManager.class); + StreamingTaskManager streamingTaskManager = Mockito.mock(StreamingTaskManager.class); + GlobalTransactionMgrIface transactionMgr = Mockito.mock(GlobalTransactionMgrIface.class); + TxnStateCallbackFactory callbackFactory = Mockito.mock(TxnStateCallbackFactory.class); + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr); + Mockito.when(env.getJobManager()).thenReturn(jobManager); + Mockito.when(jobManager.getStreamingTaskManager()).thenReturn(streamingTaskManager); + Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory); + + Assert.assertEquals(JobStatus.RUNNING, failedJob.getJobStatus()); + Assert.assertFalse(failedJob.hasReachedEnd()); + failedJob.onStreamTaskFail(failedTask); + Assert.assertEquals(JobStatus.PAUSED, failedJob.getJobStatus()); + Assert.assertEquals(0, failedJob.journalCount); + + Assert.assertEquals(JobStatus.RUNNING, succeededJob.getJobStatus()); + Assert.assertFalse(succeededJob.hasReachedEnd()); + Deencapsulation.invoke(succeededJob, "updateJobStatisticAndOffset", + new StreamingTaskTxnCommitAttachment(9001L, 1018L, 0, 0, 0, 0, 0, + succeededOffset.toSerializedJson()), false); + succeededJob.onStreamTaskSuccess(succeededTask); + Assert.assertEquals(JobStatus.FINISHED, succeededJob.getJobStatus()); + Assert.assertEquals(1, succeededJob.journalCount); + } + } + + @Test + public void testRecoveredSourceFinishesBeforeCreatingTask() throws Exception { + TestStreamingInsertJob job = newJob(new EndJdbcSourceOffsetProvider(), 1019L); + job.setJobStatus(JobStatus.PENDING); + Deencapsulation.setField(job, "runningStreamTask", null); + job.setJobRuntimeMsg("will retry"); + + try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class)) { + GlobalTransactionMgrIface transactionMgr = Mockito.mock(GlobalTransactionMgrIface.class); + TxnStateCallbackFactory callbackFactory = Mockito.mock(TxnStateCallbackFactory.class); + envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr); + Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory); + + Assert.assertTrue(job.tryFinishJob()); + } + + Assert.assertEquals(JobStatus.FINISHED, job.getJobStatus()); + Assert.assertNull(job.getRunningStreamTask()); + Assert.assertEquals("", job.getJobRuntimeMsg()); + Assert.assertEquals(1, job.journalCount); + } + + @Test + public void testCloudReplayRefreshesPersistedOffset() { + Map<String, String> properties = ImmutableMap.of("s3.ingestion_mode", "ONCE"); + S3SourceOffsetProvider provider = new S3SourceOffsetProvider(new StreamingJobProperties(properties)); + StreamingInsertJob job = new StreamingInsertJob(); + job.offsetProvider = provider; + job.setJobStatus(JobStatus.PENDING); + Deencapsulation.setField(job, "properties", properties); + Deencapsulation.setField(job, "tvfType", "s3"); + job.setOffsetProviderPersist("{\"endFile\":\"data/a.csv\"}"); + S3Offset offset = new S3Offset(); + offset.setEndFile("data/b.csv"); + offset.setLastBatch(true); + StreamingTaskTxnCommitAttachment attachment = new StreamingTaskTxnCommitAttachment( + 9001L, 1020L, 0, 0, 0, 0, 0, offset.toSerializedJson()); + + Deencapsulation.invoke(job, "updateCloudJobStatisticAndOffset", attachment, true); + + Assert.assertEquals(provider.getPersistInfo(), job.getOffsetProviderPersist()); + Assert.assertTrue(job.getOffsetProviderPersist().contains("data/b.csv")); + StreamingInsertJob recovered = GsonUtils.GSON.fromJson( + GsonUtils.GSON.toJson(job), StreamingInsertJob.class); + Assert.assertEquals(JobStatus.PENDING, recovered.getJobStatus()); + Assert.assertTrue(recovered.hasReachedEnd()); + Assert.assertFalse(recovered.hasMoreDataToConsume()); + } + @Test public void testReplayUpdatedRestoresFinalStateAndRemovesCallback() { TestStreamingInsertJob job = newJob(new JdbcSourceOffsetProvider(), 1013L); @@ -195,7 +298,7 @@ public class StreamingInsertJobOffsetPersistenceTest { Assert.assertEquals(1234L, job.getStartTimeMs()); } - private static TestStreamingInsertJob newJob(JdbcSourceOffsetProvider provider, long taskId) { + private static TestStreamingInsertJob newJob(SourceOffsetProvider provider, long taskId) { TestStreamingInsertJob job = new TestStreamingInsertJob(); Deencapsulation.setField(job, "lock", new ReentrantReadWriteLock(true)); Deencapsulation.setField(job, "jobId", 9001L); diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java new file mode 100644 index 00000000000..aedc3bd15c1 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java @@ -0,0 +1,36 @@ +// 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.job.extensions.insert.streaming; + +import org.apache.doris.job.common.JobStatus; +import org.apache.doris.job.exception.JobException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class StreamingInsertJobStatusTransitionTest { + + @Test + public void testFinishedStatusCannotBeOverwritten() { + StreamingInsertJob job = new StreamingInsertJob(); + job.setJobStatus(JobStatus.FINISHED); + + Assertions.assertThrows(JobException.class, () -> job.updateJobStatus(JobStatus.PAUSED)); + Assertions.assertEquals(JobStatus.FINISHED, job.getJobStatus()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java index ca5098e1b24..2e42c9f0c10 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java @@ -18,11 +18,15 @@ package org.apache.doris.job.extensions.insert.streaming; import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.job.common.JobStatus; import org.apache.doris.job.exception.JobException; import org.apache.doris.job.extensions.insert.InsertTask; +import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; +import com.google.common.collect.ImmutableMap; import org.junit.Assert; import org.junit.Test; @@ -30,6 +34,45 @@ import java.util.HashMap; public class StreamingJobPropertiesTest { + @Test + public void testS3OnceModeValidationAndAlter() throws Exception { + StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", " once ")); + properties.validate(); + Assert.assertTrue(properties.isS3OnceMode()); + Assert.assertThrows(AnalysisException.class, + () -> new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", "invalid")).validate()); + Assert.assertThrows(AnalysisException.class, + () -> new StreamingJobProperties( + ImmutableMap.of("s3.ingestion_mode", "ONCE", "offset", "{\"fileName\":\"a.csv\"}")) + .validate()); + + StreamingInsertJob job = new StreamingInsertJob(); + Deencapsulation.setField(job, "properties", properties.getProperties()); + AlterJobCommand alterBatch = new AlterJobCommand("job", ImmutableMap.of("s3.max_batch_files", "1"), + null, null, null, ImmutableMap.of(), ImmutableMap.of()); + Deencapsulation.invoke(alterBatch, "validateProps", job); + AlterJobCommand alterMode = new AlterJobCommand("job", ImmutableMap.of("s3.ingestion_mode", "LEXICAL"), + null, null, null, ImmutableMap.of(), ImmutableMap.of()); + Assert.assertThrows(AnalysisException.class, + () -> Deencapsulation.invoke(alterMode, "validateProps", job)); + AlterJobCommand alterOffset = new AlterJobCommand("job", + ImmutableMap.of("offset", "{\"fileName\":\"a.csv\"}"), null, null, null, ImmutableMap.of(), ImmutableMap.of()); + Assert.assertThrows(AnalysisException.class, + () -> Deencapsulation.invoke(alterOffset, "validateProps", job)); + } + + @Test + public void testS3OnceModeRestoredWithLegacyOffset() throws Exception { + StreamingInsertJob job = new StreamingInsertJob(); + Deencapsulation.setField(job, "properties", ImmutableMap.of("s3.ingestion_mode", "ONCE")); + Deencapsulation.setField(job, "tvfType", "s3"); + job.setOffsetProviderPersist("{\"endFile\":\"data/a.csv\"}"); + job.gsonPostProcess(); + job.setJobStatus(JobStatus.RUNNING); + Assert.assertFalse(job.hasMoreDataToConsume()); + Assert.assertFalse(job.hasReachedEnd()); + } + /** * Simulate FE restart: constructor is called without validate(). * Before the fix, maxIntervalSecond would be 0 when properties is non-empty, 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 new file mode 100644 index 00000000000..a2f75c9aae7 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java @@ -0,0 +1,151 @@ +// 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.job.offset.s3; + +import org.apache.doris.backup.Status; +import org.apache.doris.datasource.property.storage.StorageProperties; +import org.apache.doris.fs.FileSystemFactory; +import org.apache.doris.fs.GlobListResult; +import org.apache.doris.fs.remote.RemoteFile; +import org.apache.doris.fs.remote.RemoteFileSystem; +import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties; + +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.stubbing.Answer; + +import java.util.List; +import java.util.Map; + +public class S3SourceOffsetProviderTest { + private static final Map<String, String> TVF_PROPS = ImmutableMap.of( + "uri", "s3://bucket/data/*.csv", "s3.endpoint", "s3.us-east-1.amazonaws.com", + "s3.region", "us-east-1", "s3.access_key", "ak", "s3.secret_key", "sk"); + private static final StreamingJobProperties ONCE_PROPS = new StreamingJobProperties( + ImmutableMap.of("s3.ingestion_mode", "ONCE", "s3.max_batch_files", "1")); + + @Test + public void testOnceMarksLastBatchAndRecovers() 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); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.isNull(), + Mockito.eq(1L), Mockito.eq(1L))).thenAnswer(page("data/a.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.isNull(), + Mockito.eq(ONCE_PROPS.getS3BatchBytes()), Mockito.eq(1L))) + .thenAnswer(page("data/a.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.eq("data/a.csv"), + Mockito.eq(1L), Mockito.eq(1L))) + .thenAnswer(page("data/b.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.eq("data/a.csv"), + Mockito.eq(ONCE_PROPS.getS3BatchBytes()), Mockito.eq(1L))) + .thenAnswer(page("data/b.csv", "data/b.csv")); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.eq("data/b.csv"), + Mockito.eq(1L), Mockito.eq(1L))) + .thenReturn(new GlobListResult(Status.OK, "", "bucket", "data/")); + + S3SourceOffsetProvider provider = new S3SourceOffsetProvider(ONCE_PROPS); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + provider.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + S3Offset first = provider.getNextOffset(ONCE_PROPS, TVF_PROPS); + Assertions.assertEquals("data/a.csv", first.getEndFile()); + Assertions.assertFalse(first.isLastBatch()); + Assertions.assertEquals("data/a.csv", provider.getNextOffset(ONCE_PROPS, TVF_PROPS).getEndFile()); + provider.updateOffset(provider.deserializeOffset(first.toSerializedJson())); + Assertions.assertTrue(provider.hasMoreDataToConsume()); + S3Offset second = provider.getNextOffset(ONCE_PROPS, TVF_PROPS); + Assertions.assertEquals("data/b.csv", second.getEndFile()); + Assertions.assertTrue(second.isLastBatch()); + Assertions.assertFalse(provider.hasReachedEnd()); + // A concurrent metadata probe cannot change this task's completion decision. + provider.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(second.isLastBatch()); + Assertions.assertFalse(provider.hasReachedEnd()); + Assertions.assertEquals("data/b.csv", provider.getNextOffset(ONCE_PROPS, TVF_PROPS).getEndFile()); + provider.updateOffset(provider.deserializeOffset(second.toSerializedJson())); + Assertions.assertTrue(provider.hasReachedEnd()); + Assertions.assertFalse(provider.hasMoreDataToConsume()); + + S3SourceOffsetProvider recovered = new S3SourceOffsetProvider(ONCE_PROPS); + recovered.restoreFromPersistInfo(provider.getPersistInfo()); + Assertions.assertTrue(recovered.hasReachedEnd()); + Assertions.assertFalse(recovered.hasMoreDataToConsume()); + + recovered.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(recovered.hasReachedEnd()); + + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.eq("data/b.csv"), + Mockito.eq(1L), Mockito.eq(1L))).thenAnswer(page("data/c.csv", "data/c.csv")); + S3SourceOffsetProvider recoveredWithNewFile = new S3SourceOffsetProvider(ONCE_PROPS); + recoveredWithNewFile.restoreFromPersistInfo(provider.getPersistInfo()); + recoveredWithNewFile.fetchRemoteMeta(TVF_PROPS); + Assertions.assertTrue(recoveredWithNewFile.hasReachedEnd()); + Assertions.assertFalse(recoveredWithNewFile.hasMoreDataToConsume()); + + S3SourceOffsetProvider legacyWithNewFile = new S3SourceOffsetProvider(); + legacyWithNewFile.restoreFromPersistInfo("{\"endFile\":\"data/b.csv\"}"); + legacyWithNewFile.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(legacyWithNewFile.hasReachedEnd()); + Assertions.assertTrue(legacyWithNewFile.hasMoreDataToConsume()); + + S3SourceOffsetProvider lexical = new S3SourceOffsetProvider(); + lexical.restoreFromPersistInfo(provider.getPersistInfo()); + lexical.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(lexical.hasReachedEnd()); + Assertions.assertTrue(lexical.hasMoreDataToConsume()); + } + } + + @Test + public void testEmptyAndListingErrorRemainDistinct() 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); + Mockito.when(fs.globListWithLimit(Mockito.anyString(), Mockito.anyList(), Mockito.isNull(), + Mockito.anyLong(), Mockito.anyLong())) + .thenReturn(new GlobListResult(new Status(Status.ErrCode.COMMON_ERROR, "listing failed"))) + .thenReturn(new GlobListResult(Status.OK, "", "bucket", "data/")); + + S3SourceOffsetProvider once = new S3SourceOffsetProvider(ONCE_PROPS); + Exception listingError = Assertions.assertThrows(Exception.class, () -> once.fetchRemoteMeta(TVF_PROPS)); + Assertions.assertTrue(listingError.getMessage().contains("listing failed")); + Assertions.assertFalse(once.hasReachedEnd()); + once.fetchRemoteMeta(TVF_PROPS); + Assertions.assertFalse(once.hasReachedEnd()); + Assertions.assertTrue(once.hasMoreDataToConsume()); + RuntimeException emptyError = Assertions.assertThrows(RuntimeException.class, + () -> once.getNextOffset(ONCE_PROPS, TVF_PROPS)); + Assertions.assertTrue(emptyError.getMessage().contains("No new files found in path:")); + + S3SourceOffsetProvider lexical = new S3SourceOffsetProvider(); + Assertions.assertThrows(RuntimeException.class, () -> lexical.getNextOffset(ONCE_PROPS, TVF_PROPS)); + } + } + + private static Answer<GlobListResult> page(String key, String maxFile) { + return invocation -> { + List<RemoteFile> files = invocation.getArgument(1); + files.add(new RemoteFile("s3://bucket/" + key, true, 10, 0)); + return new GlobListResult(Status.OK, maxFile, "bucket", "data/"); + }; + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java new file mode 100644 index 00000000000..d2a5676102c --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/scheduler/StreamingTaskSchedulerTest.java @@ -0,0 +1,70 @@ +// 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.job.scheduler; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.job.common.JobStatus; +import org.apache.doris.job.extensions.insert.streaming.AbstractStreamingTask; +import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob; +import org.apache.doris.job.manager.JobManager; +import org.apache.doris.job.manager.StreamingTaskManager; + +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +public class StreamingTaskSchedulerTest { + + @Test + public void testRecoveredSourceFinishesOrContinuesAfterProbe() throws Exception { + Env env = Mockito.mock(Env.class); + JobManager<?, ?> jobManager = Mockito.mock(JobManager.class); + StreamingTaskManager taskManager = Mockito.mock(StreamingTaskManager.class); + StreamingInsertJob exhaustedJob = Mockito.mock(StreamingInsertJob.class); + AbstractStreamingTask exhaustedTask = Mockito.mock(AbstractStreamingTask.class); + StreamingInsertJob continuedJob = Mockito.mock(StreamingInsertJob.class); + AbstractStreamingTask continuedTask = Mockito.mock(AbstractStreamingTask.class); + + Mockito.when(env.getJobManager()).thenReturn(jobManager); + Mockito.when(jobManager.getStreamingTaskManager()).thenReturn(taskManager); + Mockito.when(exhaustedTask.getJobId()).thenReturn(1L); + Mockito.doReturn(exhaustedJob).when(jobManager).getJob(1L); + Mockito.when(exhaustedJob.needScheduleTask()).thenReturn(true); + Mockito.when(exhaustedJob.hasMoreDataToConsume()).thenReturn(false); + Mockito.when(exhaustedJob.hasReachedEnd()).thenReturn(true); + Mockito.when(continuedTask.getJobId()).thenReturn(2L); + Mockito.doReturn(continuedJob).when(jobManager).getJob(2L); + Mockito.when(continuedJob.needScheduleTask()).thenReturn(true); + Mockito.when(continuedJob.hasMoreDataToConsume()).thenReturn(true); + + try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class)) { + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + StreamingTaskScheduler scheduler = new StreamingTaskScheduler(); + + Deencapsulation.invoke(scheduler, "scheduleOneTask", exhaustedTask); + Mockito.verify(exhaustedJob).tryFinishJob(); + Mockito.verify(exhaustedTask, Mockito.never()).execute(); + + Deencapsulation.invoke(scheduler, "scheduleOneTask", continuedTask); + Mockito.verify(taskManager).addRunningTask(continuedTask); + Mockito.verify(continuedTask).execute(); + Mockito.verify(continuedJob, Mockito.never()).updateJobStatus(JobStatus.FINISHED); + } + } +} diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy new file mode 100644 index 00000000000..799312f3de9 --- /dev/null +++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy @@ -0,0 +1,107 @@ +// 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. + +import org.awaitility.Awaitility + +import static java.util.concurrent.TimeUnit.SECONDS + +suite("test_streaming_insert_job_s3_once") { + sql """DROP JOB IF EXISTS WHERE jobname = 'test_streaming_insert_job_s3_once'""" + sql """DROP TABLE IF EXISTS test_streaming_insert_job_s3_once_tbl FORCE""" + + sql """ + CREATE TABLE test_streaming_insert_job_s3_once_tbl ( + `c1` INT NULL, + `c2` STRING NULL, + `c3` INT NULL + ) ENGINE=OLAP + DUPLICATE KEY(`c1`) + DISTRIBUTED BY HASH(`c1`) BUCKETS 3 + PROPERTIES ("replication_allocation" = "tag.location.default: 1") + """ + + test { + sql """ + CREATE JOB test_streaming_insert_job_s3_once + PROPERTIES ( + "s3.ingestion_mode" = "ONCE", + "offset" = '{"fileName":"regression/load/data/example_0.csv"}' + ) + ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl + SELECT * FROM S3 ( + "uri" = "s3://${s3BucketName}/regression/load/data/example_[0-1].csv", + "format" = "csv", + "provider" = "${getS3Provider()}", + "column_separator" = ",", + "s3.endpoint" = "${getS3Endpoint()}", + "s3.region" = "${getS3Region()}", + "s3.access_key" = "${getS3AK()}", + "s3.secret_key" = "${getS3SK()}" + ) + """ + exception "offset is not supported when s3.ingestion_mode is ONCE" + } + + sql """ + CREATE JOB test_streaming_insert_job_s3_once + PROPERTIES ( + "s3.ingestion_mode" = "ONCE", + "s3.max_batch_files" = "1" + ) + ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl + SELECT * FROM S3 ( + "uri" = "s3://${s3BucketName}/regression/load/data/example_[0-1].csv", + "format" = "csv", + "provider" = "${getS3Provider()}", + "column_separator" = ",", + "s3.endpoint" = "${getS3Endpoint()}", + "s3.region" = "${getS3Region()}", + "s3.access_key" = "${getS3AK()}", + "s3.secret_key" = "${getS3SK()}" + ) + """ + + try { + Awaitility.await().atMost(300, SECONDS) + .pollInterval(1, SECONDS).until { + def job = sql """ + SELECT Status, SucceedTaskCount + FROM jobs("type"="insert") + WHERE Name = 'test_streaming_insert_job_s3_once' + AND ExecuteType = 'STREAMING' + """ + def rows = sql """SELECT COUNT(*) FROM test_streaming_insert_job_s3_once_tbl""" + log.info("S3 ONCE job: ${job}, row count: ${rows}") + job.size() == 1 + && job.get(0).get(0) == "FINISHED" + && job.get(0).get(1).toString() == "2" + && rows.get(0).get(0).toString() == "20" + } + } catch (Exception ex) { + def showJob = sql """ + SELECT * FROM jobs("type"="insert") + WHERE Name = 'test_streaming_insert_job_s3_once' + """ + def showTask = sql """ + SELECT * FROM tasks("type"="insert") + WHERE JobName = 'test_streaming_insert_job_s3_once' + """ + log.info("show job: " + showJob) + log.info("show task: " + showTask) + throw ex + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
