leaves12138 commented on code in PR #486:
URL: https://github.com/apache/paimon-rust/pull/486#discussion_r3544616561
##########
crates/paimon/src/table/data_evolution_reader.rs:
##########
@@ -375,6 +387,121 @@ impl DataEvolutionReader {
})
}
+ async fn finish_wide_batch(
+ &self,
+ batch: RecordBatch,
+ blob_view_lookup: Option<&BlobViewLookup>,
+ descriptor_fields: &HashSet<String>,
+ filter_before_blob_resolution: bool,
+ ) -> crate::Result<RecordBatch> {
+ let mut batch = if filter_before_blob_resolution {
+ self.filter_wide_batch(batch)?
+ } else {
+ batch
+ };
+ if filter_before_blob_resolution && batch.num_rows() == 0 {
+ return self.project_output(batch);
+ }
+
+ batch = self.resolve_blob_view_columns(batch, blob_view_lookup)?;
+ let mut batch = if !self.blob_as_descriptor &&
!descriptor_fields.is_empty() {
+ resolve_descriptor_columns(batch, descriptor_fields,
&self.file_io).await?
+ } else {
+ batch
+ };
+
+ if !filter_before_blob_resolution {
+ batch = self.filter_wide_batch(batch)?;
+ }
+ self.project_output(batch)
+ }
+
+ fn can_filter_before_blob_resolution(
+ &self,
+ resolve_blob_views: bool,
+ descriptor_fields: &HashSet<String>,
+ ) -> bool {
+ let mut transformed_fields = HashSet::new();
+ if resolve_blob_views {
+ transformed_fields.extend(self.blob_view_fields.iter().cloned());
+ }
+ if !self.blob_as_descriptor {
+ transformed_fields.extend(descriptor_fields.iter().cloned());
+ }
+ transformed_fields.is_empty()
+ || !predicates_reference_any_field(
+ &self.predicates,
+ &transformed_fields,
+ &self.table_fields,
+ )
+ }
+
+ fn blob_view_read_fields(&self) -> Vec<DataField> {
+ if !self.blob_view_resolve_enabled ||
self.blob_view_rest_env.is_none() {
+ return Vec::new();
+ }
+ self.wide_file_read_type
+ .iter()
+ .filter(|field| self.blob_view_fields.contains(field.name()))
+ .cloned()
+ .collect()
+ }
+
+ async fn preload_blob_view_lookup(
+ &self,
+ splits: &[DataSplit],
+ ) -> crate::Result<Option<BlobViewLookup>> {
+ let view_fields = self.blob_view_read_fields();
+ if view_fields.is_empty() {
+ return Ok(None);
+ }
+ let Some(rest_env) = self.blob_view_rest_env.clone() else {
+ return Ok(None);
+ };
+
+ let prescan = DataEvolutionReader::new(
+ self.file_io.clone(),
+ self.schema_manager.clone(),
+ self.table_schema_id,
+ self.table_fields.clone(),
+ view_fields.clone(),
+ Vec::new(),
Review Comment:
Java's `DataEvolutionTableRead#createBlobViewReader` applies the current
predicate to the Blob View prescan before collecting `BlobViewStruct`s. This
prescan drops the predicate, so Rust preloads/resolves every Blob View
reference in the split, including rows that will be filtered out later. That
can change behavior (for example, a filtered-out row pointing to a missing
upstream table still fails the scan) and can do unnecessary upstream reads.
Could we filter the prescan to the same final row set when the predicate is
safe to evaluate before Blob View resolution, and add a regression test for a
filtered-out invalid Blob View reference?
--
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]
