You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/12/05 09:05:39 UTC
(incubator-paimon) 12/12: [core] Introduce scan.file-creation-time-millis (#2452)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit e80c859bb32a78f8954debc90e995834b22b0bea
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Dec 5 17:03:21 2023 +0800
[core] Introduce scan.file-creation-time-millis (#2452)
---
docs/content/how-to/querying-tables.md | 14 +++++
.../shortcodes/generated/core_configuration.html | 10 +++-
.../main/java/org/apache/paimon/CoreOptions.java | 27 +++++++++-
.../paimon/operation/AbstractFileStoreScan.java | 33 +++++++-----
.../org/apache/paimon/operation/FileStoreScan.java | 2 +
.../org/apache/paimon/schema/SchemaValidation.java | 26 ++++++++-
.../table/source/AbstractInnerTableScan.java | 4 ++
.../snapshot/FileCreationTimeStartingScanner.java | 63 ++++++++++++++++++++++
.../table/source/snapshot/SnapshotReader.java | 2 +
.../table/source/snapshot/SnapshotReaderImpl.java | 6 +++
.../apache/paimon/table/system/AuditLogTable.java | 8 +++
.../flink/source/CompactorSourceBuilder.java | 1 +
.../flink/utils/MultiTablesCompactorUtil.java | 1 +
.../apache/paimon/flink/BatchFileStoreITCase.java | 28 ++++++++++
.../org/apache/paimon/flink/FlinkCatalogTest.java | 8 +--
15 files changed, 213 insertions(+), 20 deletions(-)
diff --git a/docs/content/how-to/querying-tables.md b/docs/content/how-to/querying-tables.md
index 5cbf59275..0c852192b 100644
--- a/docs/content/how-to/querying-tables.md
+++ b/docs/content/how-to/querying-tables.md
@@ -310,6 +310,20 @@ SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '2023-01-01 00:00:00' + INTERVAL
{{< /tabs >}}
+Time travel's stream read rely on snapshots, but by default, snapshot only retains data within 1 hour, which can
+prevent you from reading older incremental data. So, Paimon also provides another mode for streaming reads,
+`scan.file-creation-time-millis`, which provides a rough filtering to retain files generated after `timeMillis`.
+
+{{< tabs "file-creation-time-millis" >}}
+
+{{< tab "Flink (dynamic option)" >}}
+```sql
+SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
### Consumer ID
You can specify the `consumer-id` when streaming read table:
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 4684a1d37..a6024d99a 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -425,6 +425,12 @@ This config option does not affect the default filesystem metastore.</td>
<td>Long</td>
<td>End condition "watermark" for bounded streaming mode. Stream reading will end when a larger watermark snapshot is encountered.</td>
</tr>
+ <tr>
+ <td><h5>scan.file-creation-time-millis</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Long</td>
+ <td>After configuring this time, only the data files created after this time will be read. It is independent of snapshots, but it is imprecise filtering (depending on whether or not compaction occurs).</td>
+ </tr>
<tr>
<td><h5>scan.manifest.parallelism</h5></td>
<td style="word-wrap: break-word;">(none)</td>
@@ -441,7 +447,7 @@ This config option does not affect the default filesystem metastore.</td>
<td><h5>scan.mode</h5></td>
<td style="word-wrap: break-word;">default</td>
<td><p>Enum</p></td>
- <td>Specify the scanning behavior of the source.<br /><br />Possible values:<ul><li>"default": Determines actual startup mode according to other table properties. If "scan.timestamp-millis" is set the actual startup mode will be "from-timestamp", and if "scan.snapshot-id" or "scan.tag-name" is set the actual startup mode will be "from-snapshot". Otherwise the actual startup mode will be "latest-full".</li><li>"latest-full": For streaming sources, produces the latest snapshot [...]
+ <td>Specify the scanning behavior of the source.<br /><br />Possible values:<ul><li>"default": Determines actual startup mode according to other table properties. If "scan.timestamp-millis" is set the actual startup mode will be "from-timestamp", and if "scan.snapshot-id" or "scan.tag-name" is set the actual startup mode will be "from-snapshot". Otherwise the actual startup mode will be "latest-full".</li><li>"latest-full": For streaming sources, produces the latest snapshot [...]
</tr>
<tr>
<td><h5>scan.plan-sort-partition</h5></td>
@@ -465,7 +471,7 @@ This config option does not affect the default filesystem metastore.</td>
<td><h5>scan.timestamp-millis</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Optional timestamp used in case of "from-timestamp" scan mode.</td>
+ <td>Optional timestamp used in case of "from-timestamp" scan mode. If there is no snapshot earlier than this time, the earliest snapshot will be chosen.</td>
</tr>
<tr>
<td><h5>sequence.auto-padding</h5></td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index cdf60c0c3..7a24523e9 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -460,7 +460,17 @@ public class CoreOptions implements Serializable {
.noDefaultValue()
.withDeprecatedKeys("log.scan.timestamp-millis")
.withDescription(
- "Optional timestamp used in case of \"from-timestamp\" scan mode.");
+ "Optional timestamp used in case of \"from-timestamp\" scan mode. "
+ + "If there is no snapshot earlier than this time, the earliest snapshot will be chosen.");
+
+ public static final ConfigOption<Long> SCAN_FILE_CREATION_TIME_MILLIS =
+ key("scan.file-creation-time-millis")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "After configuring this time, only the data files created after this time will be read. "
+ + "It is independent of snapshots, but it is imprecise filtering (depending on whether "
+ + "or not compaction occurs).");
public static final ConfigOption<Long> SCAN_SNAPSHOT_ID =
key("scan.snapshot-id")
@@ -1231,6 +1241,8 @@ public class CoreOptions implements Serializable {
} else if (options.getOptional(SCAN_SNAPSHOT_ID).isPresent()
|| options.getOptional(SCAN_TAG_NAME).isPresent()) {
return StartupMode.FROM_SNAPSHOT;
+ } else if (options.getOptional(SCAN_FILE_CREATION_TIME_MILLIS).isPresent()) {
+ return StartupMode.FROM_FILE_CREATION_TIME;
} else if (options.getOptional(INCREMENTAL_BETWEEN).isPresent()
|| options.getOptional(INCREMENTAL_BETWEEN_TIMESTAMP).isPresent()) {
return StartupMode.INCREMENTAL;
@@ -1248,6 +1260,10 @@ public class CoreOptions implements Serializable {
return options.get(SCAN_TIMESTAMP_MILLIS);
}
+ public Long scanFileCreationTimeMills() {
+ return options.get(SCAN_FILE_CREATION_TIME_MILLIS);
+ }
+
public Long scanBoundedWatermark() {
return options.get(SCAN_BOUNDED_WATERMARK);
}
@@ -1513,6 +1529,11 @@ public class CoreOptions implements Serializable {
+ "For batch sources, produces a snapshot at timestamp specified by \"scan.timestamp-millis\" "
+ "but does not read new changes."),
+ FROM_FILE_CREATION_TIME(
+ "from-file-creation-time",
+ "For streaming and batch sources, produces a snapshot and filters the data files by creation time. "
+ + "For streaming sources, upon first startup, and continue to read the latest changes."),
+
FROM_SNAPSHOT(
"from-snapshot",
"For streaming sources, continuously reads changes starting from snapshot "
@@ -1811,6 +1832,10 @@ public class CoreOptions implements Serializable {
options.set(SCAN_MODE, StartupMode.FROM_TIMESTAMP);
}
+ if (options.contains(SCAN_FILE_CREATION_TIME_MILLIS) && !options.contains(SCAN_MODE)) {
+ options.set(SCAN_MODE, StartupMode.FROM_FILE_CREATION_TIME);
+ }
+
if (options.contains(SCAN_SNAPSHOT_ID) && !options.contains(SCAN_MODE)) {
options.set(SCAN_MODE, StartupMode.FROM_SNAPSHOT);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index 8bd1fa2a4..c15959b50 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -79,6 +79,7 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
private List<ManifestFileMeta> specifiedManifests = null;
private ScanMode scanMode = ScanMode.ALL;
private Filter<Integer> levelFilter = null;
+ private Long dataFileTimeMills = null;
private ManifestCacheFilter manifestCacheFilter = null;
private final Integer scanManifestParallelism;
@@ -187,6 +188,12 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
return this;
}
+ @Override
+ public FileStoreScan withDataFileTimeMills(long dataFileTimeMills) {
+ this.dataFileTimeMills = dataFileTimeMills;
+ return this;
+ }
+
@Override
public FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter) {
this.manifestCacheFilter = manifestFilter;
@@ -260,7 +267,7 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
files.parallelStream()
.filter(this::filterManifestFileMeta)
.flatMap(m -> readManifest.apply(m).stream())
- .filter(this::filterByStats)
+ .filter(this::filterUnmergedManifestEntry)
.collect(Collectors.toList());
cntEntries.getAndAdd(entryList.size());
return entryList;
@@ -295,7 +302,7 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
// however entry.bucket() was computed against the old numOfBuckets
// and thus the filtered manifest entries might be empty
// which renders the bucket check invalid
- if (filterByBucket(file) && filterByBucketSelector(file) && filterByLevel(file)) {
+ if (filterMergedManifestEntry(file)) {
files.add(file);
}
}
@@ -379,22 +386,24 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
}
/** Note: Keep this thread-safe. */
- private boolean filterByBucket(ManifestEntry entry) {
- return (bucketFilter == null || bucketFilter.test(entry.bucket()));
- }
+ private boolean filterUnmergedManifestEntry(ManifestEntry entry) {
+ if (dataFileTimeMills != null
+ && entry.file().creationTimeEpochMillis() < dataFileTimeMills) {
+ return false;
+ }
- /** Note: Keep this thread-safe. */
- private boolean filterByBucketSelector(ManifestEntry entry) {
- return bucketKeyFilter.select(entry.bucket(), entry.totalBuckets());
+ return filterByStats(entry);
}
/** Note: Keep this thread-safe. */
- private boolean filterByLevel(ManifestEntry entry) {
- return (levelFilter == null || levelFilter.test(entry.file().level()));
- }
+ protected abstract boolean filterByStats(ManifestEntry entry);
/** Note: Keep this thread-safe. */
- protected abstract boolean filterByStats(ManifestEntry entry);
+ private boolean filterMergedManifestEntry(ManifestEntry entry) {
+ return (bucketFilter == null || bucketFilter.test(entry.bucket()))
+ && bucketKeyFilter.select(entry.bucket(), entry.totalBuckets())
+ && (levelFilter == null || levelFilter.test(entry.file().level()));
+ }
/** Note: Keep this thread-safe. */
protected abstract boolean filterWholeBucketByStats(List<ManifestEntry> entries);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
index 0d11fa605..8fbd948ce 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java
@@ -61,6 +61,8 @@ public interface FileStoreScan {
FileStoreScan withLevelFilter(Filter<Integer> levelFilter);
+ FileStoreScan withDataFileTimeMills(long dataFileTimeMills);
+
FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter);
FileStoreScan withMetrics(ScanMetrics metrics);
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 4f4a1b068..c4c13b5c7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -51,6 +51,7 @@ import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
import static org.apache.paimon.CoreOptions.FULL_COMPACTION_DELTA_COMMITS;
import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN;
import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP;
+import static org.apache.paimon.CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS;
import static org.apache.paimon.CoreOptions.SCAN_MODE;
import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
import static org.apache.paimon.CoreOptions.SCAN_TAG_NAME;
@@ -236,6 +237,7 @@ public class SchemaValidation {
options,
Arrays.asList(
SCAN_SNAPSHOT_ID,
+ SCAN_FILE_CREATION_TIME_MILLIS,
SCAN_TAG_NAME,
INCREMENTAL_BETWEEN_TIMESTAMP,
INCREMENTAL_BETWEEN),
@@ -247,6 +249,7 @@ public class SchemaValidation {
options,
Arrays.asList(
SCAN_TIMESTAMP_MILLIS,
+ SCAN_FILE_CREATION_TIME_MILLIS,
INCREMENTAL_BETWEEN_TIMESTAMP,
INCREMENTAL_BETWEEN),
Arrays.asList(SCAN_SNAPSHOT_ID, SCAN_TAG_NAME));
@@ -258,7 +261,11 @@ public class SchemaValidation {
INCREMENTAL_BETWEEN_TIMESTAMP);
checkOptionsConflict(
options,
- Arrays.asList(SCAN_SNAPSHOT_ID, SCAN_TIMESTAMP_MILLIS, SCAN_TAG_NAME),
+ Arrays.asList(
+ SCAN_SNAPSHOT_ID,
+ SCAN_TIMESTAMP_MILLIS,
+ SCAN_FILE_CREATION_TIME_MILLIS,
+ SCAN_TAG_NAME),
Arrays.asList(INCREMENTAL_BETWEEN, INCREMENTAL_BETWEEN_TIMESTAMP));
} else if (options.startupMode() == CoreOptions.StartupMode.FROM_SNAPSHOT_FULL) {
checkOptionExistInMode(options, SCAN_SNAPSHOT_ID, options.startupMode());
@@ -266,12 +273,29 @@ public class SchemaValidation {
options,
Arrays.asList(
SCAN_TIMESTAMP_MILLIS,
+ SCAN_FILE_CREATION_TIME_MILLIS,
SCAN_TAG_NAME,
INCREMENTAL_BETWEEN_TIMESTAMP,
INCREMENTAL_BETWEEN),
Collections.singletonList(SCAN_SNAPSHOT_ID));
+ } else if (options.startupMode() == CoreOptions.StartupMode.FROM_FILE_CREATION_TIME) {
+ checkOptionExistInMode(
+ options,
+ SCAN_FILE_CREATION_TIME_MILLIS,
+ CoreOptions.StartupMode.FROM_FILE_CREATION_TIME);
+ checkOptionsConflict(
+ options,
+ Arrays.asList(
+ SCAN_SNAPSHOT_ID,
+ SCAN_TIMESTAMP_MILLIS,
+ SCAN_TAG_NAME,
+ INCREMENTAL_BETWEEN_TIMESTAMP,
+ INCREMENTAL_BETWEEN),
+ Collections.singletonList(SCAN_FILE_CREATION_TIME_MILLIS));
} else {
checkOptionNotExistInMode(options, SCAN_TIMESTAMP_MILLIS, options.startupMode());
+ checkOptionNotExistInMode(
+ options, SCAN_FILE_CREATION_TIME_MILLIS, options.startupMode());
checkOptionNotExistInMode(options, SCAN_SNAPSHOT_ID, options.startupMode());
checkOptionNotExistInMode(options, SCAN_TAG_NAME, options.startupMode());
checkOptionNotExistInMode(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
index 501630d80..5fad72a2a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.source.snapshot.ContinuousFromSnapshotFullStartin
import org.apache.paimon.table.source.snapshot.ContinuousFromSnapshotStartingScanner;
import org.apache.paimon.table.source.snapshot.ContinuousFromTimestampStartingScanner;
import org.apache.paimon.table.source.snapshot.ContinuousLatestStartingScanner;
+import org.apache.paimon.table.source.snapshot.FileCreationTimeStartingScanner;
import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner;
import org.apache.paimon.table.source.snapshot.FullStartingScanner;
import org.apache.paimon.table.source.snapshot.IncrementalStartingScanner;
@@ -132,6 +133,9 @@ public abstract class AbstractInnerTableScan implements InnerTableScan {
return isStreaming
? new ContinuousFromTimestampStartingScanner(snapshotManager, startupMillis)
: new StaticFromTimestampStartingScanner(snapshotManager, startupMillis);
+ case FROM_FILE_CREATION_TIME:
+ Long fileCreationTimeMills = options.scanFileCreationTimeMills();
+ return new FileCreationTimeStartingScanner(snapshotManager, fileCreationTimeMills);
case FROM_SNAPSHOT:
if (options.scanSnapshotId() != null) {
return isStreaming
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/FileCreationTimeStartingScanner.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/FileCreationTimeStartingScanner.java
new file mode 100644
index 000000000..a3a87bef2
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/FileCreationTimeStartingScanner.java
@@ -0,0 +1,63 @@
+/*
+ * 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.paimon.table.source.snapshot;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link StartingScanner} for the {@link CoreOptions.StartupMode#FROM_FILE_CREATION_TIME} startup
+ * mode.
+ */
+public class FileCreationTimeStartingScanner extends AbstractStartingScanner {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(FileCreationTimeStartingScanner.class);
+
+ private final long startupMillis;
+
+ public FileCreationTimeStartingScanner(SnapshotManager snapshotManager, long startupMillis) {
+ super(snapshotManager);
+ this.startupMillis = startupMillis;
+ }
+
+ @Override
+ public ScanMode startingScanMode() {
+ return ScanMode.ALL;
+ }
+
+ @Override
+ public Result scan(SnapshotReader snapshotReader) {
+ Long startingSnapshotId = snapshotManager.latestSnapshotId();
+ if (startingSnapshotId == null) {
+ LOG.debug("There is currently no snapshot. Waiting for snapshot generation.");
+ return new NoSnapshot();
+ }
+ return StartingScanner.fromPlan(
+ snapshotReader
+ .withMode(ScanMode.ALL)
+ .withSnapshot(startingSnapshotId)
+ .withDataFileTimeMills(startupMillis)
+ .read());
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 4af797c0a..9ccb4c49f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -57,6 +57,8 @@ public interface SnapshotReader {
SnapshotReader withLevelFilter(Filter<Integer> levelFilter);
+ SnapshotReader withDataFileTimeMills(long dataFileTimeMills);
+
SnapshotReader withBucket(int bucket);
SnapshotReader withBucketFilter(Filter<Integer> bucketFilter);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 38e996ddc..86e6f5d3e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -198,6 +198,12 @@ public class SnapshotReaderImpl implements SnapshotReader {
return this;
}
+ @Override
+ public SnapshotReader withDataFileTimeMills(long dataFileTimeMills) {
+ scan.withDataFileTimeMills(dataFileTimeMills);
+ return this;
+ }
+
@Override
public SnapshotReader withBucket(int bucket) {
scan.withBucket(bucket);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index f8e84fa72..37e5a38a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -233,16 +233,24 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
return this;
}
+ @Override
public SnapshotReader withMode(ScanMode scanMode) {
snapshotReader.withMode(scanMode);
return this;
}
+ @Override
public SnapshotReader withLevelFilter(Filter<Integer> levelFilter) {
snapshotReader.withLevelFilter(levelFilter);
return this;
}
+ @Override
+ public SnapshotReader withDataFileTimeMills(long dataFileTimeMills) {
+ snapshotReader.withDataFileTimeMills(dataFileTimeMills);
+ return this;
+ }
+
public SnapshotReader withBucket(int bucket) {
snapshotReader.withBucket(bucket);
return this;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
index fc6eda4c4..70b8a772a 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -142,6 +142,7 @@ public class CompactorSourceBuilder {
return new HashMap<String, String>() {
{
put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
+ put(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), null);
put(CoreOptions.SCAN_SNAPSHOT_ID.key(), null);
put(CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.LATEST_FULL.toString());
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
index 2fa4b3798..0fa20f192 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/MultiTablesCompactorUtil.java
@@ -50,6 +50,7 @@ public class MultiTablesCompactorUtil {
return new HashMap<String, String>() {
{
put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
+ put(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), null);
put(CoreOptions.SCAN_SNAPSHOT_ID.key(), null);
put(
CoreOptions.SCAN_MODE.key(),
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 2c2be4f12..00c7b4aa4 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -101,6 +101,18 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
time1)))
.containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
+ assertThat(
+ batchSql(
+ "SELECT * FROM T /*+ OPTIONS('scan.file-creation-time-millis'='%s') */",
+ time1))
+ .containsExactlyInAnyOrder(
+ Row.of(3, 33, 333),
+ Row.of(4, 44, 444),
+ Row.of(5, 55, 555),
+ Row.of(6, 66, 666),
+ Row.of(7, 77, 777),
+ Row.of(8, 88, 888));
+
assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='2') */"))
.containsExactlyInAnyOrder(
Row.of(1, 11, 111),
@@ -126,6 +138,15 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
Row.of(3, 33, 333),
Row.of(4, 44, 444));
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT * FROM T /*+ OPTIONS('scan.file-creation-time-millis'='%s') */",
+ time2)))
+ .containsExactlyInAnyOrder(
+ Row.of(5, 55, 555), Row.of(6, 66, 666),
+ Row.of(7, 77, 777), Row.of(8, 88, 888));
+
assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='3') */"))
.containsExactlyInAnyOrder(
Row.of(1, 11, 111),
@@ -157,6 +178,13 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
Row.of(5, 55, 555),
Row.of(6, 66, 666));
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT * FROM T /*+ OPTIONS('scan.file-creation-time-millis'='%s') */",
+ time3)))
+ .containsExactlyInAnyOrder(Row.of(7, 77, 777), Row.of(8, 88, 888));
+
assertThatThrownBy(
() ->
batchSql(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
index afae1a1ab..cdca5e7ea 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
@@ -73,6 +73,7 @@ import java.util.UUID;
import java.util.stream.Stream;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
+import static org.apache.paimon.CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS;
import static org.apache.paimon.flink.FlinkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB;
import static org.apache.paimon.flink.FlinkCatalogOptions.LOG_SYSTEM_AUTO_REGISTER;
import static org.apache.paimon.flink.FlinkConnectorOptions.LOG_SYSTEM;
@@ -289,10 +290,7 @@ public class FlinkCatalogTest {
public void testDropDb_DatabaseNotEmptyException(Map<String, String> options) throws Exception {
catalog.createDatabase(path1.getDatabaseName(), null, false);
catalog.createTable(this.path1, this.createTable(options), false);
- assertThatThrownBy(
- () -> {
- catalog.dropDatabase("db1", true, false);
- })
+ assertThatThrownBy(() -> catalog.dropDatabase("db1", true, false))
.isInstanceOf(DatabaseNotEmptyException.class)
.hasMessage("Database db1 in catalog test-catalog is not empty.");
}
@@ -636,6 +634,8 @@ public class FlinkCatalogTest {
options.put("scan.snapshot-id", "1");
} else if (mode == CoreOptions.StartupMode.FROM_TIMESTAMP) {
options.put("scan.timestamp-millis", System.currentTimeMillis() + "");
+ } else if (mode == CoreOptions.StartupMode.FROM_FILE_CREATION_TIME) {
+ options.put(SCAN_FILE_CREATION_TIME_MILLIS.key(), System.currentTimeMillis() + "");
} else if (mode == CoreOptions.StartupMode.INCREMENTAL) {
options.put("incremental-between", "2,5");
}