Allows setting custom snapshot timestamp through properties

core/src/main/java/org/apache/iceberg/RemoveSnapshots.java

@@ -96,7 +96,9 @@ public void accept(String file) {
         PropertyUtil.propertyAsLong(
             base.properties(), MAX_SNAPSHOT_AGE_MS, MAX_SNAPSHOT_AGE_MS_DEFAULT);
 
-    this.now = System.currentTimeMillis();
+    this.now =
+        ops.current()
+            .propertyAsLong(TableProperties.SNAPSHOT_TIMESTAMP, System.currentTimeMillis());
     this.defaultExpireOlderThan = now - defaultMaxSnapshotAgeMs;
     this.defaultMinNumSnapshots =
         PropertyUtil.propertyAsInt(

core/src/main/java/org/apache/iceberg/SnapshotProducer.java

@@ -28,6 +28,7 @@
 import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
 import static org.apache.iceberg.TableProperties.MANIFEST_TARGET_SIZE_BYTES;
 import static org.apache.iceberg.TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT;
+import static org.apache.iceberg.TableProperties.SNAPSHOT_TIMESTAMP;
 
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.LoadingCache;
@@ -252,11 +253,14 @@ public Snapshot apply() {
       throw new RuntimeIOException(e, "Failed to write manifest list file");
     }
 
+    final long timeToSet =
+        ops.current().propertyAsLong(SNAPSHOT_TIMESTAMP, System.currentTimeMillis());
+
     return new BaseSnapshot(
         sequenceNumber,
         snapshotId(),
         parentSnapshotId,
-        System.currentTimeMillis(),
+        timeToSet,
         operation(),
         summary(base),
         base.currentSchemaId(),

core/src/main/java/org/apache/iceberg/TableMetadata.java

@@ -1216,7 +1216,11 @@ public Builder setRef(String name, SnapshotRef ref) {
       if (SnapshotRef.MAIN_BRANCH.equals(name)) {
         this.currentSnapshotId = ref.snapshotId();
         if (lastUpdatedMillis == null) {
-          this.lastUpdatedMillis = System.currentTimeMillis();
+          this.lastUpdatedMillis =
+              Long.parseLong(
+                  properties.getOrDefault(
+                      TableProperties.SNAPSHOT_TIMESTAMP,
+                      Long.toString(System.currentTimeMillis())));
         }
 
         snapshotLog.add(new SnapshotLogEntry(lastUpdatedMillis, ref.snapshotId()));
@@ -1343,6 +1347,14 @@ public Builder setProperties(Map<String, String> updated) {
         return this;
       }
 
+      if (updated.get(TableProperties.SNAPSHOT_TIMESTAMP) != null
+          && base.lastUpdatedMillis != 0L
+          && Long.parseLong(updated.get(TableProperties.SNAPSHOT_TIMESTAMP))
+              < base.lastUpdatedMillis) {
+        throw new IllegalArgumentException(
+            "Cannot set snapshot.timestamp to a time earlier than the latest snapshot");
+      }
+
       properties.putAll(updated);
       changes.add(new MetadataUpdate.SetProperties(updated));
 
@@ -1393,7 +1405,10 @@ public TableMetadata build() {
       }
 
       if (lastUpdatedMillis == null) {
-        this.lastUpdatedMillis = System.currentTimeMillis();
+        this.lastUpdatedMillis =
+            Long.parseLong(
+                properties.getOrDefault(
+                    TableProperties.SNAPSHOT_TIMESTAMP, Long.toString(System.currentTimeMillis())));
       }
 
       // when associated with a metadata file, table metadata must have no changes so that the

core/src/main/java/org/apache/iceberg/TableProperties.java

@@ -360,4 +360,6 @@ private TableProperties() {}
 
   public static final String UPSERT_ENABLED = "write.upsert.enabled";
   public static final boolean UPSERT_ENABLED_DEFAULT = false;
+
+  public static final String SNAPSHOT_TIMESTAMP = "snapshot.timestamp";
 }