From 324e349bfede45047497d51eb02247a656293e28 Mon Sep 17 00:00:00 2001 From: yanghua Date: Fri, 12 Jun 2026 18:09:10 +0800 Subject: [PATCH 1/5] feat: support cleanup explain for python and java --- java/lance-jni/src/blocking_dataset.rs | 175 ++++++++++++++++-- java/src/main/java/org/lance/Dataset.java | 16 ++ .../lance/cleanup/CleanupCandidateFile.java | 50 +++++ .../org/lance/cleanup/CleanupExplanation.java | 72 +++++++ .../org/lance/cleanup/CleanupFileKind.java | 43 +++++ .../cleanup/CleanupReferencedBranch.java | 39 ++++ java/src/test/java/org/lance/CleanupTest.java | 33 ++++ python/python/lance/dataset.py | 59 ++++++ python/python/lance/lance/__init__.pyi | 33 +++- python/python/tests/test_dataset.py | 39 ++++ python/src/dataset.rs | 166 ++++++++++++++--- python/src/dataset/cleanup.rs | 54 +++++- python/src/lib.rs | 7 +- 13 files changed, 741 insertions(+), 45 deletions(-) create mode 100644 java/src/main/java/org/lance/cleanup/CleanupCandidateFile.java create mode 100644 java/src/main/java/org/lance/cleanup/CleanupExplanation.java create mode 100644 java/src/main/java/org/lance/cleanup/CleanupFileKind.java create mode 100644 java/src/main/java/org/lance/cleanup/CleanupReferencedBranch.java diff --git a/java/lance-jni/src/blocking_dataset.rs b/java/lance-jni/src/blocking_dataset.rs index 1d06f3eed87..ea99f9d7659 100644 --- a/java/lance-jni/src/blocking_dataset.rs +++ b/java/lance-jni/src/blocking_dataset.rs @@ -29,7 +29,10 @@ use jni::sys::{jboolean, jint}; use jni::sys::{jbyteArray, jlong}; use jni::{JNIEnv, objects::JObject}; use lance::dataset::builder::DatasetBuilder; -use lance::dataset::cleanup::{CleanupPolicy, RemovalStats}; +use lance::dataset::cleanup::{ + CleanupCandidateFile, CleanupExplanation, CleanupFileKind, CleanupPolicy, + CleanupReferencedBranch, RemovalStats, +}; use lance::dataset::optimize::{CompactionOptions as RustCompactionOptions, compact_files}; use lance::dataset::refs::{Ref, TagContents}; use lance::dataset::statistics::{DataStatistics, DatasetStatisticsExt}; @@ -414,6 +417,10 @@ impl BlockingDataset { Ok(RT.block_on(self.inner.cleanup_with_policy(policy))?) } + pub fn explain_cleanup_with_policy(&self, policy: CleanupPolicy) -> Result { + Ok(RT.block_on(self.inner.cleanup(policy).explain())?) + } + pub fn close(&self) {} } @@ -3063,6 +3070,46 @@ fn inner_cleanup_with_policy<'local>( jdataset: JObject, jpolicy: JObject, ) -> Result> { + let policy = extract_cleanup_policy(env, &jpolicy)?; + + let stats = { + let mut dataset = + unsafe { env.get_rust_field::<_, _, BlockingDataset>(jdataset, NATIVE_DATASET) }?; + dataset.cleanup_with_policy(policy) + }?; + + cleanup_stats_to_java(env, stats) +} + +#[unsafe(no_mangle)] +pub extern "system" fn Java_org_lance_Dataset_nativeExplainCleanupWithPolicy<'local>( + mut env: JNIEnv<'local>, + jdataset: JObject, + jpolicy: JObject, +) -> JObject<'local> { + ok_or_throw!( + env, + inner_explain_cleanup_with_policy(&mut env, jdataset, jpolicy) + ) +} + +fn inner_explain_cleanup_with_policy<'local>( + env: &mut JNIEnv<'local>, + jdataset: JObject, + jpolicy: JObject, +) -> Result> { + let policy = extract_cleanup_policy(env, &jpolicy)?; + + let explanation = { + let dataset = + unsafe { env.get_rust_field::<_, _, BlockingDataset>(jdataset, NATIVE_DATASET) }?; + dataset.explain_cleanup_with_policy(policy) + }?; + + cleanup_explanation_to_java(env, explanation) +} + +fn extract_cleanup_policy(env: &mut JNIEnv<'_>, jpolicy: &JObject) -> Result { let before_ts_millis = env.get_optional_u64_from_method(&jpolicy, "getBeforeTimestampMillis")?; let before_timestamp = before_ts_millis.map(|millis| { @@ -3092,22 +3139,21 @@ fn inner_cleanup_with_policy<'local>( let delete_rate_limit = env.get_optional_u64_from_method(&jpolicy, "getDeleteRateLimit")?; - let policy = CleanupPolicy { + Ok(CleanupPolicy { before_timestamp, before_version, delete_unverified, error_if_tagged_old_versions, clean_referenced_branches, delete_rate_limit, - }; - - let stats = { - let mut dataset = - unsafe { env.get_rust_field::<_, _, BlockingDataset>(jdataset, NATIVE_DATASET) }?; - dataset.cleanup_with_policy(policy) - }?; + }) +} - let jstats = env.new_object( +fn cleanup_stats_to_java<'local>( + env: &mut JNIEnv<'local>, + stats: RemovalStats, +) -> Result> { + Ok(env.new_object( "org/lance/cleanup/RemovalStats", "(JJJJJJ)V", &[ @@ -3118,9 +3164,114 @@ fn inner_cleanup_with_policy<'local>( JValue::Long(stats.index_files_removed as i64), JValue::Long(stats.deletion_files_removed as i64), ], - )?; + )?) +} - Ok(jstats) +fn cleanup_file_kind_to_java(kind: CleanupFileKind) -> &'static str { + match kind { + CleanupFileKind::Manifest => "manifest", + CleanupFileKind::Data => "data", + CleanupFileKind::Transaction => "transaction", + CleanupFileKind::Index => "index", + CleanupFileKind::Deletion => "deletion", + CleanupFileKind::TemporaryManifest => "temporary_manifest", + } +} + +fn cleanup_candidate_files_to_java<'local>( + env: &mut JNIEnv<'local>, + files: Vec, +) -> Result> { + let list = env.new_object("java/util/ArrayList", "()V", &[])?; + for file in files { + let path = env.new_string(file.path)?; + let kind = env.new_string(cleanup_file_kind_to_java(file.kind))?; + let candidate = env.new_object( + "org/lance/cleanup/CleanupCandidateFile", + "(Ljava/lang/String;Ljava/lang/String;ZJ)V", + &[ + JValue::Object(&path), + JValue::Object(&kind), + JValue::Bool(file.unverified as jboolean), + JValue::Long(file.size_bytes as i64), + ], + )?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&candidate)], + )?; + } + Ok(list) +} + +fn cleanup_referenced_branches_to_java<'local>( + env: &mut JNIEnv<'local>, + branches: Vec, +) -> Result> { + let list = env.new_object("java/util/ArrayList", "()V", &[])?; + for branch in branches { + let name = env.new_string(branch.name)?; + let referenced_branch = env.new_object( + "org/lance/cleanup/CleanupReferencedBranch", + "(Ljava/lang/String;JZ)V", + &[ + JValue::Object(&name), + JValue::Long(branch.referenced_version as i64), + JValue::Bool(branch.cleanup_candidate as jboolean), + ], + )?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&referenced_branch)], + )?; + } + Ok(list) +} + +fn cleanup_warnings_to_java<'local>( + env: &mut JNIEnv<'local>, + warnings: Vec, +) -> Result> { + let list = env.new_object("java/util/ArrayList", "()V", &[])?; + for warning in warnings { + let warning = env.new_string(warning)?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&warning)], + )?; + } + Ok(list) +} + +fn cleanup_explanation_to_java<'local>( + env: &mut JNIEnv<'local>, + explanation: CleanupExplanation, +) -> Result> { + let stats = cleanup_stats_to_java(env, explanation.stats)?; + let candidate_files = cleanup_candidate_files_to_java(env, explanation.candidate_files)?; + let referenced_branches = + cleanup_referenced_branches_to_java(env, explanation.referenced_branches)?; + let warnings = cleanup_warnings_to_java(env, explanation.warnings)?; + + Ok(env.new_object( + "org/lance/cleanup/CleanupExplanation", + "(JLorg/lance/cleanup/RemovalStats;Ljava/util/List;ZJLjava/util/List;Ljava/util/List;)V", + &[ + JValue::Long(explanation.read_version as i64), + JValue::Object(&stats), + JValue::Object(&candidate_files), + JValue::Bool(explanation.candidate_files_truncated as jboolean), + JValue::Long(explanation.candidate_file_limit as i64), + JValue::Object(&referenced_branches), + JValue::Object(&warnings), + ], + )?) } ////////////////////////////// diff --git a/java/src/main/java/org/lance/Dataset.java b/java/src/main/java/org/lance/Dataset.java index 23341283861..5c5c5e73b86 100644 --- a/java/src/main/java/org/lance/Dataset.java +++ b/java/src/main/java/org/lance/Dataset.java @@ -13,6 +13,7 @@ */ package org.lance; +import org.lance.cleanup.CleanupExplanation; import org.lance.cleanup.CleanupPolicy; import org.lance.cleanup.RemovalStats; import org.lance.compaction.CompactionOptions; @@ -2156,4 +2157,19 @@ public RemovalStats cleanupWithPolicy(CleanupPolicy policy) { } private native RemovalStats nativeCleanupWithPolicy(CleanupPolicy policy); + + /** + * Explain cleanup based on a specified policy without deleting files. + * + * @param policy cleanup policy + * @return cleanup explanation + */ + public CleanupExplanation explainCleanupWithPolicy(CleanupPolicy policy) { + try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) { + Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed"); + return nativeExplainCleanupWithPolicy(policy); + } + } + + private native CleanupExplanation nativeExplainCleanupWithPolicy(CleanupPolicy policy); } diff --git a/java/src/main/java/org/lance/cleanup/CleanupCandidateFile.java b/java/src/main/java/org/lance/cleanup/CleanupCandidateFile.java new file mode 100644 index 00000000000..a7925a746af --- /dev/null +++ b/java/src/main/java/org/lance/cleanup/CleanupCandidateFile.java @@ -0,0 +1,50 @@ +/* + * Licensed 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.lance.cleanup; + +/** A file that cleanup identified as removable. */ +public class CleanupCandidateFile { + private final String path; + private final CleanupFileKind kind; + private final boolean unverified; + private final long sizeBytes; + + public CleanupCandidateFile(String path, String kind, boolean unverified, long sizeBytes) { + this(path, CleanupFileKind.fromRustString(kind), unverified, sizeBytes); + } + + public CleanupCandidateFile( + String path, CleanupFileKind kind, boolean unverified, long sizeBytes) { + this.path = path; + this.kind = kind; + this.unverified = unverified; + this.sizeBytes = sizeBytes; + } + + public String getPath() { + return path; + } + + public CleanupFileKind getKind() { + return kind; + } + + public boolean isUnverified() { + return unverified; + } + + public long getSizeBytes() { + return sizeBytes; + } +} diff --git a/java/src/main/java/org/lance/cleanup/CleanupExplanation.java b/java/src/main/java/org/lance/cleanup/CleanupExplanation.java new file mode 100644 index 00000000000..320d555c3b5 --- /dev/null +++ b/java/src/main/java/org/lance/cleanup/CleanupExplanation.java @@ -0,0 +1,72 @@ +/* + * Licensed 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.lance.cleanup; + +import java.util.List; + +/** Read-only explanation of what cleanup would remove. */ +public class CleanupExplanation { + private final long readVersion; + private final RemovalStats stats; + private final List candidateFiles; + private final boolean candidateFilesTruncated; + private final long candidateFileLimit; + private final List referencedBranches; + private final List warnings; + + public CleanupExplanation( + long readVersion, + RemovalStats stats, + List candidateFiles, + boolean candidateFilesTruncated, + long candidateFileLimit, + List referencedBranches, + List warnings) { + this.readVersion = readVersion; + this.stats = stats; + this.candidateFiles = candidateFiles; + this.candidateFilesTruncated = candidateFilesTruncated; + this.candidateFileLimit = candidateFileLimit; + this.referencedBranches = referencedBranches; + this.warnings = warnings; + } + + public long getReadVersion() { + return readVersion; + } + + public RemovalStats getStats() { + return stats; + } + + public List getCandidateFiles() { + return candidateFiles; + } + + public boolean isCandidateFilesTruncated() { + return candidateFilesTruncated; + } + + public long getCandidateFileLimit() { + return candidateFileLimit; + } + + public List getReferencedBranches() { + return referencedBranches; + } + + public List getWarnings() { + return warnings; + } +} diff --git a/java/src/main/java/org/lance/cleanup/CleanupFileKind.java b/java/src/main/java/org/lance/cleanup/CleanupFileKind.java new file mode 100644 index 00000000000..695e778fc76 --- /dev/null +++ b/java/src/main/java/org/lance/cleanup/CleanupFileKind.java @@ -0,0 +1,43 @@ +/* + * Licensed 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.lance.cleanup; + +/** Kind of file identified by cleanup. */ +public enum CleanupFileKind { + MANIFEST, + DATA, + TRANSACTION, + INDEX, + DELETION, + TEMPORARY_MANIFEST; + + public static CleanupFileKind fromRustString(String value) { + switch (value) { + case "manifest": + return MANIFEST; + case "data": + return DATA; + case "transaction": + return TRANSACTION; + case "index": + return INDEX; + case "deletion": + return DELETION; + case "temporary_manifest": + return TEMPORARY_MANIFEST; + default: + throw new IllegalArgumentException("Unknown cleanup file kind: " + value); + } + } +} diff --git a/java/src/main/java/org/lance/cleanup/CleanupReferencedBranch.java b/java/src/main/java/org/lance/cleanup/CleanupReferencedBranch.java new file mode 100644 index 00000000000..47fae25cd5a --- /dev/null +++ b/java/src/main/java/org/lance/cleanup/CleanupReferencedBranch.java @@ -0,0 +1,39 @@ +/* + * Licensed 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.lance.cleanup; + +/** A branch that references the current branch lineage. */ +public class CleanupReferencedBranch { + private final String name; + private final long referencedVersion; + private final boolean cleanupCandidate; + + public CleanupReferencedBranch(String name, long referencedVersion, boolean cleanupCandidate) { + this.name = name; + this.referencedVersion = referencedVersion; + this.cleanupCandidate = cleanupCandidate; + } + + public String getName() { + return name; + } + + public long getReferencedVersion() { + return referencedVersion; + } + + public boolean isCleanupCandidate() { + return cleanupCandidate; + } +} diff --git a/java/src/test/java/org/lance/CleanupTest.java b/java/src/test/java/org/lance/CleanupTest.java index f287f1d0f0a..434dcb8fe3e 100644 --- a/java/src/test/java/org/lance/CleanupTest.java +++ b/java/src/test/java/org/lance/CleanupTest.java @@ -13,6 +13,7 @@ */ package org.lance; +import org.lance.cleanup.CleanupExplanation; import org.lance.cleanup.CleanupPolicy; import org.lance.cleanup.RemovalStats; @@ -53,6 +54,38 @@ public void testCleanupBeforeVersion(@TempDir Path tempDir) { } } + @Test + public void testExplainCleanupBeforeVersion(@TempDir Path tempDir) { + String datasetPath = tempDir.resolve("test_dataset_for_cleanup").toString(); + try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + TestUtils.SimpleTestDataset testDataset = + new TestUtils.SimpleTestDataset(allocator, datasetPath); + + testDataset.createEmptyDataset().close(); + + testDataset.write(1, 10).close(); + testDataset.write(2, 10).close(); + + try (Dataset dataset = testDataset.write(3, 10)) { + CleanupPolicy policy = CleanupPolicy.builder().withBeforeVersion(3L).build(); + CleanupExplanation explanation = dataset.explainCleanupWithPolicy(policy); + + assertEquals(2L, explanation.getStats().getOldVersions()); + assertEquals(2L, explanation.getStats().getTransactionFilesRemoved()); + assertTrue(explanation.getStats().getBytesRemoved() > 0); + assertTrue(explanation.getReadVersion() > 0); + assertTrue(explanation.getCandidateFiles().size() > 0); + assertTrue(explanation.getReferencedBranches().isEmpty()); + + List versions = dataset.listVersions(); + assertEquals(4, versions.size()); + + RemovalStats stats = dataset.cleanupWithPolicy(policy); + assertEquals(explanation.getStats().getOldVersions(), stats.getOldVersions()); + } + } + } + @Test public void testCleanupBeforeTimestamp(@TempDir Path tempDir) throws Exception { String datasetPath = tempDir.resolve("test_dataset_for_cleanup").toString(); diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py index dae72b88b1c..ddede62e307 100644 --- a/python/python/lance/dataset.py +++ b/python/python/lance/dataset.py @@ -51,6 +51,7 @@ from .fragment import DataFile, FragmentMetadata, LanceFragment from .indices import IndexConfig, IndexSegment, SupportedDistributedIndices from .lance import ( + CleanupExplanation, CleanupStats, Compaction, CompactionMetrics, @@ -2976,6 +2977,64 @@ def cleanup_old_versions( delete_rate_limit, ) + def explain_cleanup_old_versions( + self, + older_than: Optional[timedelta] = None, + retain_versions: Optional[int] = None, + *, + delete_unverified: bool = False, + error_if_tagged_old_versions: bool = True, + delete_rate_limit: Optional[int] = None, + include_files: bool = False, + max_files: int = 1000, + ) -> CleanupExplanation: + """ + Explain what :meth:`cleanup_old_versions` would remove without deleting files. + + Parameters + ---------- + + older_than: timedelta, optional + Only versions older than this would be removed. If ``older_than`` and + ``retain_versions`` are not specified, this will default to two weeks. + + retain_versions: int, optional + Retain the last N versions of the dataset. + + delete_unverified: bool, default False + Include unverified files that cleanup would remove when this is set. + + error_if_tagged_old_versions: bool, default True + If set to `True`, an exception will be raised if any tagged versions + match the parameters. Otherwise, tagged versions will be ignored. + + delete_rate_limit: int, optional + Accepted for parity with :meth:`cleanup_old_versions`; no deletes are + issued by explain. + + include_files: bool, default False + If `True`, include candidate files in the explanation up to + ``max_files`` entries. Aggregate stats always include all candidates. + + max_files: int, default 1000 + Maximum number of candidate files to include when ``include_files`` + is `True`. + """ + if older_than is None and retain_versions is None: + older_than = timedelta(days=14) + if max_files <= 0: + raise ValueError("max_files must be positive") + + return self._ds.explain_cleanup_old_versions( + td_to_micros(older_than) if older_than else None, + retain_versions, + delete_unverified, + error_if_tagged_old_versions, + delete_rate_limit, + include_files, + max_files, + ) + def _prepare_scalar_index_request( self, column: Union[str, List[str]], diff --git a/python/python/lance/lance/__init__.pyi b/python/python/lance/lance/__init__.pyi index 38d82738063..58830d7ccc0 100644 --- a/python/python/lance/lance/__init__.pyi +++ b/python/python/lance/lance/__init__.pyi @@ -95,6 +95,26 @@ class CleanupStats: index_files_removed: int deletion_files_removed: int +class CleanupCandidateFile: + path: str + kind: str + unverified: bool + size_bytes: int + +class CleanupReferencedBranch: + name: str + referenced_version: int + cleanup_candidate: bool + +class CleanupExplanation: + read_version: int + stats: CleanupStats + candidate_files: List[CleanupCandidateFile] + candidate_files_truncated: bool + candidate_file_limit: int + referenced_branches: List[CleanupReferencedBranch] + warnings: List[str] + class LanceFileWriter: def __init__( self, @@ -342,11 +362,22 @@ class _Dataset: def restore(self): ... def cleanup_old_versions( self, - older_than_micros: int, + older_than_micros: Optional[int] = None, + retain_versions: Optional[int] = None, delete_unverified: Optional[bool] = None, error_if_tagged_old_versions: Optional[bool] = None, delete_rate_limit: Optional[int] = None, ) -> CleanupStats: ... + def explain_cleanup_old_versions( + self, + older_than_micros: Optional[int] = None, + retain_versions: Optional[int] = None, + delete_unverified: Optional[bool] = None, + error_if_tagged_old_versions: Optional[bool] = None, + delete_rate_limit: Optional[int] = None, + include_files: bool = False, + max_files: int = 1000, + ) -> CleanupExplanation: ... def get_version(self, tag: str) -> int: ... # Tag operations def tags(self) -> Dict[str, Tag]: ... diff --git a/python/python/tests/test_dataset.py b/python/python/tests/test_dataset.py index 89bd78b82c8..c6155950136 100644 --- a/python/python/tests/test_dataset.py +++ b/python/python/tests/test_dataset.py @@ -1377,6 +1377,45 @@ def test_cleanup_old_versions(tmp_path): assert stats.old_versions == 1 +def test_explain_cleanup_old_versions(tmp_path): + table = pa.Table.from_pydict({"a": range(100), "b": range(100)}) + base_dir = tmp_path / "test" + lance.write_dataset(table, base_dir) + time.sleep(0.1) + moment = datetime.now() + lance.write_dataset(table, base_dir, mode="overwrite") + + dataset = lance.dataset(base_dir) + before_versions = len(dataset.versions()) + + explanation = dataset.explain_cleanup_old_versions( + older_than=(datetime.now() - moment), + include_files=True, + max_files=1000, + ) + + assert explanation.read_version == dataset.version + assert explanation.stats.bytes_removed > 0 + assert explanation.stats.old_versions == 1 + assert explanation.candidate_files + assert not explanation.candidate_files_truncated + assert len(dataset.versions()) == before_versions + + summary = dataset.explain_cleanup_old_versions(older_than=(datetime.now() - moment)) + assert summary.stats.old_versions == explanation.stats.old_versions + assert summary.candidate_files == [] + + with pytest.raises(ValueError, match="max_files must be positive"): + dataset.explain_cleanup_old_versions( + older_than=(datetime.now() - moment), + max_files=0, + ) + + stats = dataset.cleanup_old_versions(older_than=(datetime.now() - moment)) + assert stats.bytes_removed == explanation.stats.bytes_removed + assert stats.old_versions == explanation.stats.old_versions + + def test_cleanup_error_when_tagged_old_versions(tmp_path): table = pa.Table.from_pydict({"a": range(100), "b": range(100)}) base_dir = tmp_path / "test" diff --git a/python/src/dataset.rs b/python/src/dataset.rs index 8bfa81aeae4..38f5598ef9a 100644 --- a/python/src/dataset.rs +++ b/python/src/dataset.rs @@ -37,7 +37,7 @@ use pyo3::{ use uuid::Uuid; use lance::dataset::AutoCleanupParams; -use lance::dataset::cleanup::CleanupPolicyBuilder; +use lance::dataset::cleanup::{CleanupFileKind, CleanupPolicyBuilder}; use lance::dataset::refs::{Ref, TagContents}; use lance::dataset::scanner::{ AggregateExpr, ColumnOrdering, DatasetRecordBatchStream, ExecutionStatsCallback, @@ -105,7 +105,9 @@ use crate::utils::PyLance; use crate::{LanceReader, Scanner}; use lance::io::commit::namespace_manifest::LanceNamespaceExternalManifestStore; -use self::cleanup::CleanupStats; +use self::cleanup::{ + CleanupCandidateFile, CleanupExplanation, CleanupReferencedBranch, CleanupStats, +}; use self::commit::PyCommitLock; use self::io_stats::IoStats; @@ -682,6 +684,89 @@ pub struct Dataset { pub(crate) ds: Arc, } +impl Dataset { + async fn cleanup_policy( + &self, + older_than_micros: Option, + retain_versions: Option, + delete_unverified: Option, + error_if_tagged_old_versions: Option, + delete_rate_limit: Option, + ) -> lance_core::Result { + let mut builder = CleanupPolicyBuilder::default(); + if let Some(v) = older_than_micros { + let older_than = Duration::microseconds(v); + builder = builder.before_timestamp(Utc::now() - older_than); + } + if let Some(v) = retain_versions { + builder = builder.retain_n_versions(self.ds.as_ref(), v).await?; + } + if let Some(v) = delete_unverified { + builder = builder.delete_unverified(v); + } + if let Some(v) = error_if_tagged_old_versions { + builder = builder.error_if_tagged_old_versions(v); + } + if let Some(v) = delete_rate_limit { + builder = builder.delete_rate_limit(v)?; + } + Ok(builder.build()) + } +} + +fn cleanup_stats(stats: lance::dataset::cleanup::RemovalStats) -> CleanupStats { + CleanupStats { + bytes_removed: stats.bytes_removed, + old_versions: stats.old_versions, + data_files_removed: stats.data_files_removed, + transaction_files_removed: stats.transaction_files_removed, + index_files_removed: stats.index_files_removed, + deletion_files_removed: stats.deletion_files_removed, + } +} + +fn cleanup_file_kind(kind: CleanupFileKind) -> &'static str { + match kind { + CleanupFileKind::Manifest => "manifest", + CleanupFileKind::Data => "data", + CleanupFileKind::Transaction => "transaction", + CleanupFileKind::Index => "index", + CleanupFileKind::Deletion => "deletion", + CleanupFileKind::TemporaryManifest => "temporary_manifest", + } +} + +fn cleanup_explanation( + explanation: lance::dataset::cleanup::CleanupExplanation, +) -> CleanupExplanation { + CleanupExplanation { + read_version: explanation.read_version, + stats: cleanup_stats(explanation.stats), + candidate_files: explanation + .candidate_files + .into_iter() + .map(|file| CleanupCandidateFile { + path: file.path, + kind: cleanup_file_kind(file.kind).to_string(), + unverified: file.unverified, + size_bytes: file.size_bytes, + }) + .collect(), + candidate_files_truncated: explanation.candidate_files_truncated, + candidate_file_limit: explanation.candidate_file_limit, + referenced_branches: explanation + .referenced_branches + .into_iter() + .map(|branch| CleanupReferencedBranch { + name: branch.name, + referenced_version: branch.referenced_version, + cleanup_candidate: branch.cleanup_candidate, + }) + .collect(), + warnings: explanation.warnings, + } +} + #[pymethods] impl Dataset { #[allow(clippy::too_many_arguments)] @@ -1860,37 +1945,60 @@ impl Dataset { error_if_tagged_old_versions: Option, delete_rate_limit: Option, ) -> PyResult { - let cleanup_stats = rt() + let stats = rt() .block_on(None, async { - let mut builder = CleanupPolicyBuilder::default(); - if let Some(v) = older_than_micros { - let older_than = Duration::microseconds(v); - builder = builder.before_timestamp(Utc::now() - older_than); - } - if let Some(v) = retain_versions { - builder = builder.retain_n_versions(self.ds.as_ref(), v).await?; - } - if let Some(v) = delete_unverified { - builder = builder.delete_unverified(v); - } - if let Some(v) = error_if_tagged_old_versions { - builder = builder.error_if_tagged_old_versions(v); - } - if let Some(v) = delete_rate_limit { - builder = builder.delete_rate_limit(v)?; - } + let policy = self + .cleanup_policy( + older_than_micros, + retain_versions, + delete_unverified, + error_if_tagged_old_versions, + delete_rate_limit, + ) + .await?; + self.ds.cleanup_with_policy(policy).await + })? + .map_err(|err: lance::Error| PyIOError::new_err(err.to_string()))?; + Ok(cleanup_stats(stats)) + } - self.ds.cleanup_with_policy(builder.build()).await + /// Explain cleanup old versions from the dataset without deleting files + #[pyo3(signature = (older_than_micros = None, retain_versions = None, delete_unverified = None, error_if_tagged_old_versions = None, delete_rate_limit = None, include_files = false, max_files = 1000))] + fn explain_cleanup_old_versions( + &self, + older_than_micros: Option, + retain_versions: Option, + delete_unverified: Option, + error_if_tagged_old_versions: Option, + delete_rate_limit: Option, + include_files: bool, + max_files: usize, + ) -> PyResult { + let explanation = rt() + .block_on(None, async { + let policy = self + .cleanup_policy( + older_than_micros, + retain_versions, + delete_unverified, + error_if_tagged_old_versions, + delete_rate_limit, + ) + .await?; + self.ds + .cleanup(policy) + .with_max_candidate_files(max_files) + .explain() + .await })? .map_err(|err: lance::Error| PyIOError::new_err(err.to_string()))?; - Ok(CleanupStats { - bytes_removed: cleanup_stats.bytes_removed, - old_versions: cleanup_stats.old_versions, - data_files_removed: cleanup_stats.data_files_removed, - transaction_files_removed: cleanup_stats.transaction_files_removed, - index_files_removed: cleanup_stats.index_files_removed, - deletion_files_removed: cleanup_stats.deletion_files_removed, - }) + let mut explanation = cleanup_explanation(explanation); + if !include_files { + explanation.candidate_files.clear(); + explanation.candidate_files_truncated = false; + explanation.warnings.clear(); + } + Ok(explanation) } fn tags_ordered(self_: PyRef<'_, Self>, order: Option) -> PyResult> { diff --git a/python/src/dataset/cleanup.rs b/python/src/dataset/cleanup.rs index 4f1655e6df6..fb675124c4e 100644 --- a/python/src/dataset/cleanup.rs +++ b/python/src/dataset/cleanup.rs @@ -14,8 +14,8 @@ use pyo3::{pyclass, pymethods}; -#[pyclass(get_all)] -#[derive(Debug)] +#[pyclass(get_all, skip_from_py_object)] +#[derive(Clone, Debug)] pub struct CleanupStats { pub bytes_removed: u64, pub old_versions: u64, @@ -31,3 +31,53 @@ impl CleanupStats { format!("{self:?}") } } + +#[pyclass(get_all, skip_from_py_object)] +#[derive(Clone, Debug)] +pub struct CleanupCandidateFile { + pub path: String, + pub kind: String, + pub unverified: bool, + pub size_bytes: u64, +} + +#[pymethods] +impl CleanupCandidateFile { + fn __repr__(&self) -> String { + format!("{self:?}") + } +} + +#[pyclass(get_all, skip_from_py_object)] +#[derive(Clone, Debug)] +pub struct CleanupReferencedBranch { + pub name: String, + pub referenced_version: u64, + pub cleanup_candidate: bool, +} + +#[pymethods] +impl CleanupReferencedBranch { + fn __repr__(&self) -> String { + format!("{self:?}") + } +} + +#[pyclass(get_all, skip_from_py_object)] +#[derive(Clone, Debug)] +pub struct CleanupExplanation { + pub read_version: u64, + pub stats: CleanupStats, + pub candidate_files: Vec, + pub candidate_files_truncated: bool, + pub candidate_file_limit: usize, + pub referenced_branches: Vec, + pub warnings: Vec, +} + +#[pymethods] +impl CleanupExplanation { + fn __repr__(&self) -> String { + format!("{self:?}") + } +} diff --git a/python/src/lib.rs b/python/src/lib.rs index cf29b26c46a..ce9103af788 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -38,7 +38,9 @@ use datafusion_ffi::table_provider::FFI_TableProvider; #[cfg(feature = "datagen")] use datagen::register_datagen; use dataset::blob::LanceBlobFile; -use dataset::cleanup::CleanupStats; +use dataset::cleanup::{ + CleanupCandidateFile, CleanupExplanation, CleanupReferencedBranch, CleanupStats, +}; use dataset::io_stats::IoStats; use dataset::optimize::{ PyCompaction, PyCompactionMetrics, PyCompactionPlan, PyCompactionTask, PyRewriteResult, @@ -265,6 +267,9 @@ fn lance(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; From 953d8a83a28ccf24300c12d5d293ba6f4cd49895 Mon Sep 17 00:00:00 2001 From: yanghua Date: Fri, 12 Jun 2026 18:19:48 +0800 Subject: [PATCH 2/5] feat: support cleanup explain for python and java --- .../main/java/org/lance/CleanupOperation.java | 54 +++++++++++++++++++ java/src/main/java/org/lance/Dataset.java | 26 ++++++--- java/src/test/java/org/lance/CleanupTest.java | 5 +- python/python/lance/__init__.py | 8 +++ 4 files changed, 84 insertions(+), 9 deletions(-) create mode 100644 java/src/main/java/org/lance/CleanupOperation.java diff --git a/java/src/main/java/org/lance/CleanupOperation.java b/java/src/main/java/org/lance/CleanupOperation.java new file mode 100644 index 00000000000..1ac534096cf --- /dev/null +++ b/java/src/main/java/org/lance/CleanupOperation.java @@ -0,0 +1,54 @@ +/* + * Licensed 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.lance; + +import org.lance.cleanup.CleanupExplanation; +import org.lance.cleanup.CleanupPolicy; +import org.lance.cleanup.RemovalStats; + +import org.apache.arrow.util.Preconditions; + +/** + * A cleanup operation with separate read-only explain and destructive execute actions. + * + *

This is not a deletion plan. Calling {@link #execute()} re-evaluates the current dataset and + * reference state before deleting files. + */ +public class CleanupOperation { + private final Dataset dataset; + private final CleanupPolicy policy; + + CleanupOperation(Dataset dataset, CleanupPolicy policy) { + this.dataset = Preconditions.checkNotNull(dataset, "dataset cannot be null"); + this.policy = Preconditions.checkNotNull(policy, "policy cannot be null"); + } + + /** + * Explain what cleanup would remove without deleting files. + * + * @return cleanup explanation + */ + public CleanupExplanation explain() { + return dataset.explainCleanup(policy); + } + + /** + * Execute cleanup, re-evaluating the current dataset and reference state before deleting files. + * + * @return removal stats + */ + public RemovalStats execute() { + return dataset.executeCleanup(policy); + } +} diff --git a/java/src/main/java/org/lance/Dataset.java b/java/src/main/java/org/lance/Dataset.java index 5c5c5e73b86..6c4c6b3dd7e 100644 --- a/java/src/main/java/org/lance/Dataset.java +++ b/java/src/main/java/org/lance/Dataset.java @@ -2143,6 +2143,20 @@ public Dataset shallowClone(String targetPath, Ref ref, Map stor private native Dataset nativeShallowClone( String targetPath, Ref ref, Optional> storageOptions); + /** + * Create a cleanup operation for the specified policy. + * + *

Use {@link CleanupOperation#explain()} to inspect what cleanup would remove without deleting + * files, or {@link CleanupOperation#execute()} to perform cleanup. + * + * @param policy cleanup policy + * @return cleanup operation + */ + public CleanupOperation cleanup(CleanupPolicy policy) { + Preconditions.checkNotNull(policy, "policy cannot be null"); + return new CleanupOperation(this, policy); + } + /** * Cleanup dataset based on a specified policy. * @@ -2150,6 +2164,10 @@ private native Dataset nativeShallowClone( * @return removal stats */ public RemovalStats cleanupWithPolicy(CleanupPolicy policy) { + return cleanup(policy).execute(); + } + + RemovalStats executeCleanup(CleanupPolicy policy) { try (LockManager.WriteLock writeLock = lockManager.acquireWriteLock()) { Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed"); return nativeCleanupWithPolicy(policy); @@ -2158,13 +2176,7 @@ public RemovalStats cleanupWithPolicy(CleanupPolicy policy) { private native RemovalStats nativeCleanupWithPolicy(CleanupPolicy policy); - /** - * Explain cleanup based on a specified policy without deleting files. - * - * @param policy cleanup policy - * @return cleanup explanation - */ - public CleanupExplanation explainCleanupWithPolicy(CleanupPolicy policy) { + CleanupExplanation explainCleanup(CleanupPolicy policy) { try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) { Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed"); return nativeExplainCleanupWithPolicy(policy); diff --git a/java/src/test/java/org/lance/CleanupTest.java b/java/src/test/java/org/lance/CleanupTest.java index 434dcb8fe3e..960ba4c7f2f 100644 --- a/java/src/test/java/org/lance/CleanupTest.java +++ b/java/src/test/java/org/lance/CleanupTest.java @@ -68,7 +68,8 @@ public void testExplainCleanupBeforeVersion(@TempDir Path tempDir) { try (Dataset dataset = testDataset.write(3, 10)) { CleanupPolicy policy = CleanupPolicy.builder().withBeforeVersion(3L).build(); - CleanupExplanation explanation = dataset.explainCleanupWithPolicy(policy); + CleanupOperation cleanup = dataset.cleanup(policy); + CleanupExplanation explanation = cleanup.explain(); assertEquals(2L, explanation.getStats().getOldVersions()); assertEquals(2L, explanation.getStats().getTransactionFilesRemoved()); @@ -80,7 +81,7 @@ public void testExplainCleanupBeforeVersion(@TempDir Path tempDir) { List versions = dataset.listVersions(); assertEquals(4, versions.size()); - RemovalStats stats = dataset.cleanupWithPolicy(policy); + RemovalStats stats = cleanup.execute(); assertEquals(explanation.getStats().getOldVersions(), stats.getOldVersions()); } } diff --git a/python/python/lance/__init__.py b/python/python/lance/__init__.py index f58b169a47a..7b25d542fe0 100644 --- a/python/python/lance/__init__.py +++ b/python/python/lance/__init__.py @@ -27,6 +27,10 @@ ) from .fragment import FragmentMetadata, LanceFragment from .lance import ( + CleanupCandidateFile, + CleanupExplanation, + CleanupReferencedBranch, + CleanupStats, DatasetBasePath, FFILanceTableProvider, ScanStatistics, @@ -70,6 +74,10 @@ "BlobFile", "blob_array", "blob_field", + "CleanupCandidateFile", + "CleanupExplanation", + "CleanupReferencedBranch", + "CleanupStats", "DatasetBasePath", "DataStatistics", "FieldStatistics", From af7c0b9b49f4c77532e9ae463ee1ee9cd37c9a66 Mon Sep 17 00:00:00 2001 From: yanghua Date: Fri, 12 Jun 2026 20:44:49 +0800 Subject: [PATCH 3/5] fix clippy issue --- python/src/dataset.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/python/src/dataset.rs b/python/src/dataset.rs index 38f5598ef9a..2f852cf11a4 100644 --- a/python/src/dataset.rs +++ b/python/src/dataset.rs @@ -1963,6 +1963,7 @@ impl Dataset { } /// Explain cleanup old versions from the dataset without deleting files + #[allow(clippy::too_many_arguments)] #[pyo3(signature = (older_than_micros = None, retain_versions = None, delete_unverified = None, error_if_tagged_old_versions = None, delete_rate_limit = None, include_files = false, max_files = 1000))] fn explain_cleanup_old_versions( &self, From b65994c8765c7d86184fdf2cef23cae67adc5e97 Mon Sep 17 00:00:00 2001 From: yanghua Date: Fri, 19 Jun 2026 13:13:57 +0800 Subject: [PATCH 4/5] address review comments --- java/lance-jni/src/blocking_dataset.rs | 125 +++++++++++------- .../main/java/org/lance/CleanupOperation.java | 22 ++- java/src/main/java/org/lance/Dataset.java | 7 +- java/src/test/java/org/lance/CleanupTest.java | 33 +++++ 4 files changed, 138 insertions(+), 49 deletions(-) diff --git a/java/lance-jni/src/blocking_dataset.rs b/java/lance-jni/src/blocking_dataset.rs index ea99f9d7659..fafab4fc739 100644 --- a/java/lance-jni/src/blocking_dataset.rs +++ b/java/lance-jni/src/blocking_dataset.rs @@ -417,8 +417,16 @@ impl BlockingDataset { Ok(RT.block_on(self.inner.cleanup_with_policy(policy))?) } - pub fn explain_cleanup_with_policy(&self, policy: CleanupPolicy) -> Result { - Ok(RT.block_on(self.inner.cleanup(policy).explain())?) + pub fn explain_cleanup_with_policy( + &self, + policy: CleanupPolicy, + max_candidate_files: Option, + ) -> Result { + let mut op = self.inner.cleanup(policy); + if let Some(limit) = max_candidate_files { + op = op.with_max_candidate_files(limit); + } + Ok(RT.block_on(op.explain())?) } pub fn close(&self) {} @@ -3086,10 +3094,11 @@ pub extern "system" fn Java_org_lance_Dataset_nativeExplainCleanupWithPolicy<'lo mut env: JNIEnv<'local>, jdataset: JObject, jpolicy: JObject, + jmax_candidate_files: JObject, ) -> JObject<'local> { ok_or_throw!( env, - inner_explain_cleanup_with_policy(&mut env, jdataset, jpolicy) + inner_explain_cleanup_with_policy(&mut env, jdataset, jpolicy, jmax_candidate_files) ) } @@ -3097,13 +3106,27 @@ fn inner_explain_cleanup_with_policy<'local>( env: &mut JNIEnv<'local>, jdataset: JObject, jpolicy: JObject, + jmax_candidate_files: JObject, ) -> Result> { let policy = extract_cleanup_policy(env, &jpolicy)?; + let max_candidate_files = env + .get_optional(&jmax_candidate_files, |env, inner| { + Ok(env.call_method(inner, "longValue", "()J", &[])?.j()?) + })? + .map(|v| { + usize::try_from(v).map_err(|e| { + Error::input_error(format!( + "maxCandidateFiles must be a non-negative usize value, got {}: {:?}", + v, e + )) + }) + }) + .transpose()?; let explanation = { let dataset = unsafe { env.get_rust_field::<_, _, BlockingDataset>(jdataset, NATIVE_DATASET) }?; - dataset.explain_cleanup_with_policy(policy) + dataset.explain_cleanup_with_policy(policy, max_candidate_files) }?; cleanup_explanation_to_java(env, explanation) @@ -3183,25 +3206,31 @@ fn cleanup_candidate_files_to_java<'local>( files: Vec, ) -> Result> { let list = env.new_object("java/util/ArrayList", "()V", &[])?; + // Wrap each iteration in a local frame so the temporary path/kind/candidate + // references do not accumulate on the JNI local reference table for large + // explanations (default limit is 1000 candidate files, but users can raise it). for file in files { - let path = env.new_string(file.path)?; - let kind = env.new_string(cleanup_file_kind_to_java(file.kind))?; - let candidate = env.new_object( - "org/lance/cleanup/CleanupCandidateFile", - "(Ljava/lang/String;Ljava/lang/String;ZJ)V", - &[ - JValue::Object(&path), - JValue::Object(&kind), - JValue::Bool(file.unverified as jboolean), - JValue::Long(file.size_bytes as i64), - ], - )?; - env.call_method( - &list, - "add", - "(Ljava/lang/Object;)Z", - &[JValue::Object(&candidate)], - )?; + env.with_local_frame(8, |env| { + let path = env.new_string(file.path)?; + let kind = env.new_string(cleanup_file_kind_to_java(file.kind))?; + let candidate = env.new_object( + "org/lance/cleanup/CleanupCandidateFile", + "(Ljava/lang/String;Ljava/lang/String;ZJ)V", + &[ + JValue::Object(&path), + JValue::Object(&kind), + JValue::Bool(file.unverified as jboolean), + JValue::Long(file.size_bytes as i64), + ], + )?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&candidate)], + )?; + Ok::<(), Error>(()) + })?; } Ok(list) } @@ -3212,22 +3241,25 @@ fn cleanup_referenced_branches_to_java<'local>( ) -> Result> { let list = env.new_object("java/util/ArrayList", "()V", &[])?; for branch in branches { - let name = env.new_string(branch.name)?; - let referenced_branch = env.new_object( - "org/lance/cleanup/CleanupReferencedBranch", - "(Ljava/lang/String;JZ)V", - &[ - JValue::Object(&name), - JValue::Long(branch.referenced_version as i64), - JValue::Bool(branch.cleanup_candidate as jboolean), - ], - )?; - env.call_method( - &list, - "add", - "(Ljava/lang/Object;)Z", - &[JValue::Object(&referenced_branch)], - )?; + env.with_local_frame(8, |env| { + let name = env.new_string(branch.name)?; + let referenced_branch = env.new_object( + "org/lance/cleanup/CleanupReferencedBranch", + "(Ljava/lang/String;JZ)V", + &[ + JValue::Object(&name), + JValue::Long(branch.referenced_version as i64), + JValue::Bool(branch.cleanup_candidate as jboolean), + ], + )?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&referenced_branch)], + )?; + Ok::<(), Error>(()) + })?; } Ok(list) } @@ -3238,13 +3270,16 @@ fn cleanup_warnings_to_java<'local>( ) -> Result> { let list = env.new_object("java/util/ArrayList", "()V", &[])?; for warning in warnings { - let warning = env.new_string(warning)?; - env.call_method( - &list, - "add", - "(Ljava/lang/Object;)Z", - &[JValue::Object(&warning)], - )?; + env.with_local_frame(4, |env| { + let warning = env.new_string(warning)?; + env.call_method( + &list, + "add", + "(Ljava/lang/Object;)Z", + &[JValue::Object(&warning)], + )?; + Ok::<(), Error>(()) + })?; } Ok(list) } diff --git a/java/src/main/java/org/lance/CleanupOperation.java b/java/src/main/java/org/lance/CleanupOperation.java index 1ac534096cf..1f64473f43e 100644 --- a/java/src/main/java/org/lance/CleanupOperation.java +++ b/java/src/main/java/org/lance/CleanupOperation.java @@ -19,6 +19,8 @@ import org.apache.arrow.util.Preconditions; +import java.util.Optional; + /** * A cleanup operation with separate read-only explain and destructive execute actions. * @@ -28,19 +30,37 @@ public class CleanupOperation { private final Dataset dataset; private final CleanupPolicy policy; + private Optional maxCandidateFiles = Optional.empty(); CleanupOperation(Dataset dataset, CleanupPolicy policy) { this.dataset = Preconditions.checkNotNull(dataset, "dataset cannot be null"); this.policy = Preconditions.checkNotNull(policy, "policy cannot be null"); } + /** + * Set the maximum number of candidate files included in the {@link #explain()} result. + * + *

Defaults to 1000 if not set. The aggregate {@link RemovalStats} returned by {@link + * #explain()} still account for all files that would be removed regardless of this limit; only + * the per-file {@code candidateFiles} list is truncated. + * + * @param maxCandidateFiles maximum number of candidate files to include; must be positive + * @return this operation for chaining + */ + public CleanupOperation withMaxCandidateFiles(long maxCandidateFiles) { + Preconditions.checkArgument( + maxCandidateFiles > 0, "maxCandidateFiles must be positive, got %s", maxCandidateFiles); + this.maxCandidateFiles = Optional.of(maxCandidateFiles); + return this; + } + /** * Explain what cleanup would remove without deleting files. * * @return cleanup explanation */ public CleanupExplanation explain() { - return dataset.explainCleanup(policy); + return dataset.explainCleanup(policy, maxCandidateFiles); } /** diff --git a/java/src/main/java/org/lance/Dataset.java b/java/src/main/java/org/lance/Dataset.java index 6c4c6b3dd7e..8f79d7cbaea 100644 --- a/java/src/main/java/org/lance/Dataset.java +++ b/java/src/main/java/org/lance/Dataset.java @@ -2176,12 +2176,13 @@ RemovalStats executeCleanup(CleanupPolicy policy) { private native RemovalStats nativeCleanupWithPolicy(CleanupPolicy policy); - CleanupExplanation explainCleanup(CleanupPolicy policy) { + CleanupExplanation explainCleanup(CleanupPolicy policy, Optional maxCandidateFiles) { try (LockManager.ReadLock readLock = lockManager.acquireReadLock()) { Preconditions.checkArgument(nativeDatasetHandle != 0, "Dataset is closed"); - return nativeExplainCleanupWithPolicy(policy); + return nativeExplainCleanupWithPolicy(policy, maxCandidateFiles); } } - private native CleanupExplanation nativeExplainCleanupWithPolicy(CleanupPolicy policy); + private native CleanupExplanation nativeExplainCleanupWithPolicy( + CleanupPolicy policy, Optional maxCandidateFiles); } diff --git a/java/src/test/java/org/lance/CleanupTest.java b/java/src/test/java/org/lance/CleanupTest.java index 960ba4c7f2f..5fc8ceeaa3f 100644 --- a/java/src/test/java/org/lance/CleanupTest.java +++ b/java/src/test/java/org/lance/CleanupTest.java @@ -152,6 +152,39 @@ public void testCleanupTaggedVersion(@TempDir Path tempDir) throws Exception { } } + @Test + public void testExplainCleanupWithMaxCandidateFiles(@TempDir Path tempDir) { + String datasetPath = tempDir.resolve("test_dataset_for_cleanup").toString(); + try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + TestUtils.SimpleTestDataset testDataset = + new TestUtils.SimpleTestDataset(allocator, datasetPath); + + testDataset.createEmptyDataset().close(); + + testDataset.write(1, 10).close(); + testDataset.write(2, 10).close(); + + try (Dataset dataset = testDataset.write(3, 10)) { + CleanupPolicy policy = CleanupPolicy.builder().withBeforeVersion(3L).build(); + CleanupExplanation full = dataset.cleanup(policy).explain(); + assertTrue(full.getCandidateFiles().size() > 1); + assertEquals(1000L, full.getCandidateFileLimit()); + + CleanupExplanation truncated = dataset.cleanup(policy).withMaxCandidateFiles(1L).explain(); + assertEquals(1L, truncated.getCandidateFileLimit()); + assertEquals(1, truncated.getCandidateFiles().size()); + assertTrue(truncated.isCandidateFilesTruncated()); + assertTrue(!truncated.getWarnings().isEmpty()); + // Aggregate stats stay accurate even when the per-file list is truncated. + assertEquals(full.getStats().getOldVersions(), truncated.getStats().getOldVersions()); + + Assertions.assertThrows( + IllegalArgumentException.class, + () -> dataset.cleanup(policy).withMaxCandidateFiles(0L)); + } + } + } + @Test public void testCleanupWithRateLimit(@TempDir Path tempDir) throws Exception { String datasetPath = tempDir.resolve("test_dataset_for_cleanup").toString(); From ece8ff820251da5bb984b54e9392d9abe3406f03 Mon Sep 17 00:00:00 2001 From: yanghua Date: Fri, 19 Jun 2026 13:26:41 +0800 Subject: [PATCH 5/5] fix clippy issue --- java/lance-jni/src/blocking_dataset.rs | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/java/lance-jni/src/blocking_dataset.rs b/java/lance-jni/src/blocking_dataset.rs index fafab4fc739..caf837b371a 100644 --- a/java/lance-jni/src/blocking_dataset.rs +++ b/java/lance-jni/src/blocking_dataset.rs @@ -3133,34 +3133,33 @@ fn inner_explain_cleanup_with_policy<'local>( } fn extract_cleanup_policy(env: &mut JNIEnv<'_>, jpolicy: &JObject) -> Result { - let before_ts_millis = - env.get_optional_u64_from_method(&jpolicy, "getBeforeTimestampMillis")?; + let before_ts_millis = env.get_optional_u64_from_method(jpolicy, "getBeforeTimestampMillis")?; let before_timestamp = before_ts_millis.map(|millis| { let st = UNIX_EPOCH + Duration::from_millis(millis); DateTime::::from(st) }); - let before_version = env.get_optional_u64_from_method(&jpolicy, "getBeforeVersion")?; + let before_version = env.get_optional_u64_from_method(jpolicy, "getBeforeVersion")?; let delete_unverified = env - .get_optional_from_method(&jpolicy, "getDeleteUnverified", |env, obj| { + .get_optional_from_method(jpolicy, "getDeleteUnverified", |env, obj| { Ok(env.call_method(obj, "booleanValue", "()Z", &[])?.z()?) })? .unwrap_or(false); let error_if_tagged_old_versions = env - .get_optional_from_method(&jpolicy, "getErrorIfTaggedOldVersions", |env, obj| { + .get_optional_from_method(jpolicy, "getErrorIfTaggedOldVersions", |env, obj| { Ok(env.call_method(obj, "booleanValue", "()Z", &[])?.z()?) })? .unwrap_or(true); let clean_referenced_branches = env - .get_optional_from_method(&jpolicy, "getCleanReferencedBranches", |env, obj| { + .get_optional_from_method(jpolicy, "getCleanReferencedBranches", |env, obj| { Ok(env.call_method(obj, "booleanValue", "()Z", &[])?.z()?) })? .unwrap_or(false); - let delete_rate_limit = env.get_optional_u64_from_method(&jpolicy, "getDeleteRateLimit")?; + let delete_rate_limit = env.get_optional_u64_from_method(jpolicy, "getDeleteRateLimit")?; Ok(CleanupPolicy { before_timestamp,