JingsongLi commented on code in PR #591:
URL: https://github.com/apache/paimon-rust/pull/591#discussion_r3726191174
##########
crates/paimon/src/table/format_table_scan.rs:
##########
@@ -178,19 +205,81 @@ impl<'a> FormatTableScan<'a> {
Ok(roots)
}
+ async fn catalog_managed_scan_roots(
+ &self,
+ rest_env: &RESTEnv,
+ table_path: &str,
+ partition_keys: &[String],
+ partition_fields: &[DataField],
+ managed_options: &LoadedFormatTablePartitionOptions,
+ ) -> crate::Result<Vec<ScanRoot>> {
+ let only_value_in_path = managed_options.only_value_in_path;
+ let partition_paths =
+ FormatTablePartitionPaths::new(partition_keys.iter().cloned(),
only_value_in_path);
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let default_partition_name = core_options.partition_default_name();
+ // Ask the catalog only for the partitions the filter can reach.
Downloading every
+ // registration of a table with many partitions is what dominates
planning time,
+ // and the local match below still decides what is actually scanned.
+ let pattern = match &self.partition_filter {
+ Some(filter) => {
+ let leading_values = leading_equality_partition_values(
+ filter,
+ partition_fields,
+ default_partition_name,
+ core_options.legacy_partition_name(),
+ )?;
+ partition_paths.name_prefix_pattern(&leading_values)
Review Comment:
[P1] Avoid lossy name-pattern pushdown for typed partitions
This formats typed equality literals canonically before building the raw
partitionNamePattern. MSCK preserves valid raw spellings such as active=TRUE or
month=01, but predicates active = true / month = 1 produce exact patterns
active=true / month=1. A catalog that honors the pattern removes those
registrations before the local typed check runs, so the query silently loses
rows. Please restrict name-pattern pushdown to types with a unique raw spelling
(at least CHAR/VARCHAR), or use a typed-filter/list-all fallback, and add raw
TRUE/01 scan tests.
##########
crates/paimon/src/table/format_table_scan.rs:
##########
@@ -65,17 +71,25 @@ impl<'a> FormatTableScan<'a> {
async fn plan_inner(&self, trace: Option<&mut ScanTrace>) ->
crate::Result<Plan> {
let core_options = CoreOptions::new(self.table.schema().options());
- let format_extension =
supported_format_table_extension(core_options.file_format())?;
+ let managed_options = self
+ .table
+ .rest_env()
+ .and_then(RESTEnv::catalog_managed_partition_options);
+ let file_format = managed_options
+ .map(|options| options.file_format.as_str())
+ .unwrap_or_else(|| core_options.file_format());
+ let format_extension = supported_format_table_extension(file_format)?;
let schema_id = self.table.schema().id();
- let table_path = core_options
- .path()
+ let table_path = managed_options
+ .map(|options| options.table_path.as_str())
+ .or_else(|| core_options.path())
.unwrap_or_else(|| self.table.location())
.trim_end_matches('/')
.to_string();
let partition_fields = self.table.schema().partition_fields();
let mut splits = Vec::new();
- for scan_root in self.scan_roots(&core_options, &table_path)? {
+ for scan_root in self.scan_roots(&core_options, &table_path).await? {
Review Comment:
[P2] Bound partition-directory listing concurrency
This awaits one recursive object-store listing per registered partition
inside a plain loop. An unfiltered 10k-partition table therefore pays roughly
10k list latencies serially. Java uses configurable bounded concurrency through
format-table.scan.list-parallelism (default 64). Please use bounded async
concurrency while preserving deterministic final sorting and fail-fast behavior.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -1833,6 +1908,262 @@ impl SQLContext {
ok_result(&self.ctx)
}
+ /// Unregister catalog-managed partitions and then delete their
directories.
+ ///
+ /// A specification that fixes only some of the partition keys expands to
every
+ /// registered partition it matches, the way Java Format Tables behave.
The keys need
+ /// not be a leading prefix, so `hh = '10'` alone is a valid
specification. One catalog
+ /// listing serves the whole statement, however many specifications it
carries.
+ async fn drop_catalog_managed_partitions(
+ &self,
+ catalog: &Arc<dyn Catalog>,
+ identifier: &Identifier,
+ table: &paimon::Table,
+ requests: &[(&[SqlExpr], bool)],
+ ) -> DFResult<DataFrame> {
+ ensure_catalog_managed_format_table(table, "ALTER TABLE DROP
PARTITION")?;
+ let partition_key_count = table.schema().partition_keys().len();
+ let mut requested = Vec::with_capacity(requests.len());
+ for (expressions, ignore_if_not_exists) in requests {
+ let spec = parse_format_partition_spec(expressions, table, false)?;
+ let normalized = normalize_catalog_partition_spec(&spec, table,
false)?;
+ requested.push((
+ spec.len() == partition_key_count,
+ spec,
+ normalized,
+ *ignore_if_not_exists,
+ ));
+ }
+
+ let registered = catalog
+ .list_partitions(identifier)
Review Comment:
[P2] Use list-by-names for complete DROP specs
Even a complete exact DROP PARTITION calls list_partitions and materializes
every registration. Java resolves complete specs through batched
/partitions/list-by-names and reserves a full traversal for partial specs.
Please implement the corresponding REST/Catalog method and use it for complete
requests; this avoids O(all partitions) network and memory for the common
exact-drop case.
##########
crates/paimon/src/table/format_table_scan.rs:
##########
@@ -178,19 +205,81 @@ impl<'a> FormatTableScan<'a> {
Ok(roots)
}
+ async fn catalog_managed_scan_roots(
+ &self,
+ rest_env: &RESTEnv,
+ table_path: &str,
+ partition_keys: &[String],
+ partition_fields: &[DataField],
+ managed_options: &LoadedFormatTablePartitionOptions,
+ ) -> crate::Result<Vec<ScanRoot>> {
+ let only_value_in_path = managed_options.only_value_in_path;
+ let partition_paths =
+ FormatTablePartitionPaths::new(partition_keys.iter().cloned(),
only_value_in_path);
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let default_partition_name = core_options.partition_default_name();
+ // Ask the catalog only for the partitions the filter can reach.
Downloading every
+ // registration of a table with many partitions is what dominates
planning time,
+ // and the local match below still decides what is actually scanned.
+ let pattern = match &self.partition_filter {
+ Some(filter) => {
+ let leading_values = leading_equality_partition_values(
+ filter,
+ partition_fields,
+ default_partition_name,
+ core_options.legacy_partition_name(),
+ )?;
+ partition_paths.name_prefix_pattern(&leading_values)
+ }
+ None => None,
+ };
+ let partitions = rest_env
+ .api()
+ .list_partitions_by_name_pattern(rest_env.identifier(),
pattern.as_deref())
Review Comment:
[P2] Push the complete partition predicate to REST
Only a leading-equality name pattern is sent. Filters without such a prefix
(for example, hour = 10 for keys (dt, hour), ranges, or starts-with) download
the whole registry in 1,000-row pages before local filtering. Java uses paged
/partitions/list-by-filter with the serialized predicate and still rechecks
locally. Please add the compatible request/path/API with a safe fallback.
##########
crates/paimon/src/catalog/rest/rest_catalog.rs:
##########
@@ -373,16 +375,64 @@ impl Catalog for RESTCatalog {
))
}
+ async fn create_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ if partition_specs.is_empty() {
+ return Ok(());
+ }
+ if !ignore_if_exists {
+ return self
+ .api
+ .create_partitions(identifier, partition_specs, false)
+ .await
+ .map_err(|error| map_rest_error_for_create_partitions(error,
identifier));
+ }
+
+ for batch in partition_specs.chunks(PARTITION_BATCH_SIZE) {
+ self.api
+ .create_partitions(identifier, batch.to_vec(), true)
+ .await
+ .map_err(|error| map_rest_error_for_create_partitions(error,
identifier))?;
+ }
+ Ok(())
+ }
+
+ async fn drop_partitions(
+ &self,
+ identifier: &Identifier,
+ partition_specs: Vec<HashMap<String, String>>,
+ ) -> Result<()> {
+ if partition_specs.is_empty() {
+ return Ok(());
+ }
+ for batch in partition_specs.chunks(PARTITION_BATCH_SIZE) {
Review Comment:
[P2] Preserve ordinary-table DROP semantics
RESTCatalog::drop_partitions always calls the metadata-only REST endpoint.
Java only does that for catalog-managed Format Tables; for ordinary Paimon
tables it loads the table and commits truncatePartitions. A public Rust Catalog
caller can therefore get an error or success without the expected
partition-data change. Please branch on has_catalog_managed_partitions(): keep
REST unregistering for managed Format Tables and use a table commit for
ordinary tables, or explicitly reject non-managed calls until implemented.
##########
crates/integrations/datafusion/src/sql_context.rs:
##########
@@ -1833,6 +1908,262 @@ impl SQLContext {
ok_result(&self.ctx)
}
+ /// Unregister catalog-managed partitions and then delete their
directories.
+ ///
+ /// A specification that fixes only some of the partition keys expands to
every
+ /// registered partition it matches, the way Java Format Tables behave.
The keys need
+ /// not be a leading prefix, so `hh = '10'` alone is a valid
specification. One catalog
+ /// listing serves the whole statement, however many specifications it
carries.
+ async fn drop_catalog_managed_partitions(
+ &self,
+ catalog: &Arc<dyn Catalog>,
+ identifier: &Identifier,
+ table: &paimon::Table,
+ requests: &[(&[SqlExpr], bool)],
+ ) -> DFResult<DataFrame> {
+ ensure_catalog_managed_format_table(table, "ALTER TABLE DROP
PARTITION")?;
+ let partition_key_count = table.schema().partition_keys().len();
+ let mut requested = Vec::with_capacity(requests.len());
+ for (expressions, ignore_if_not_exists) in requests {
+ let spec = parse_format_partition_spec(expressions, table, false)?;
+ let normalized = normalize_catalog_partition_spec(&spec, table,
false)?;
+ requested.push((
+ spec.len() == partition_key_count,
+ spec,
+ normalized,
+ *ignore_if_not_exists,
+ ));
+ }
+
+ let registered = catalog
+ .list_partitions(identifier)
+ .await
+ .map_err(to_datafusion_error)?
+ .into_iter()
+ .map(|partition| {
+ let normalized =
normalize_catalog_partition_spec(&partition.spec, table, true)?;
+ Ok((partition.spec, normalized))
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+
+ let core_options = CoreOptions::new(table.schema().options());
+ let partition_paths = FormatTablePartitionPaths::new(
+ table.schema().partition_keys().iter().cloned(),
+ core_options.format_table_partition_only_value_in_path(),
+ );
+ let table_path = table.location().trim_end_matches('/');
+ let mut selected: Vec<(HashMap<String, String>, String)> = Vec::new();
+ let mut selected_paths = HashSet::new();
+ for (complete, spec, normalized, ignore_if_not_exists) in &requested {
+ // A specification that is registered verbatim drops exactly that
registration,
+ // so catalog values that only differ before normalization stay
distinguishable.
+ let verbatim = registered.iter().any(|(registered, _)| registered
== spec);
+ let mut matches = 0usize;
+ for (registered_spec, registered_normalized) in ®istered {
+ let selects = if verbatim {
+ registered_spec == spec
+ } else {
+ normalized
+ .iter()
+ .all(|(key, value)| registered_normalized.get(key) ==
Some(value))
Review Comment:
[P1] Preserve raw partition identity during partial DROP
Partial specifications are matched after converting catalog values to typed
Datum values. A table partitioned by (year STRING, month INT) can legitimately
contain raw registrations such as {year=2025, month=01} and {year=2026,
month=1} after MSCK, because repair preserves directory spelling. DROP
PARTITION (month=1) has no verbatim match because the request is partial, so
both values normalize to Int(1) and both directories are unregistered and
deleted. Java compares partial specs using the raw catalog values, deleting
only month=1. Please match partial specs against raw strings and add a
mixed-spelling regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
