How to merge multiple data sources and deduplicate based on certain fields?

### Is your feature request related to a problem or challenge?

I currently have two data sources, one stored in Parquet format and the other in memory. I need to implement a scan function. I tried using UnionExec, but it's obviously not working, especially when using aggregation functions like count. Maybe I should use SortPreservingMergeExec, but there are too few examples of this function. I would appreciate it if you could add an example that includes multiple data sources, as these sources may contain duplicate data, and I would be happy to see an example of deduplication based on multiple fields.
```
async fn scan(
&self,
state: &dyn Session,
projection: Option<&Vec>,
filters: &[Expr],
limit: Option,
) -> Result> {
// convert filters like [`a = 1`, `b = 2`] to a single filter like `a = 1 AND b = 2`
let predicate = self.filters_to_predicate(state, filters)?;

// Now, we invoke the analysis code to perform the range analysis
let df_schema = DFSchema::try_from(self.schema())?;

let boundaries = ExprBoundaries::try_new_unbounded(&self.schema())?;

let analysis_result = analyze(
&predicate,
AnalysisContext::new(boundaries),
df_schema.as_ref(),
)?;

// In this example, we use the PruningPredicate's literal guarantees to
// analyze the predicate. In a real system, using
// `PruningPredicate::prune` would likely be easier to do.
let pruning_predicate =
PruningPredicate::try_new(Arc::clone(&predicate), self.schema().clone())?;

debug!("pruning_predicate:{:?}", pruning_predicate);

// The PruningPredicate's guarantees must all be satisfied in order for
// the predicate to possibly evaluate to true.
let guarantees = pruning_predicate.literal_guarantees();

debug!("guarantees:{:?}", guarantees);

let object_store_url = ObjectStoreUrl::parse("file://")?;
let mut file_scan_config = FileScanConfig::new(object_store_url, self.schema())
.with_projection(projection.cloned())
.with_limit(limit);

let mut points = vec![];

for expr in &analysis_result.boundaries {
if expr.column.name() == "show_time" {
let lower = expr.interval.lower().clone().to_array().unwrap();
let lower = lower
.as_any()
.downcast_ref::()
.unwrap();

let upper = expr.interval.upper().clone().to_array().unwrap();
let upper = upper
.as_any()
.downcast_ref::()
.unwrap();

debug!(
"{:?} lower:{:?} upper:{:?}",
expr.column,
lower.value(0),
upper.value(0)
);

let start = DateTime::::from_timestamp_millis(lower.value(0)).unwrap();
let end = DateTime::::from_timestamp_millis(upper.value(0)).unwrap();

let p = self.query_with_time(start, end).await;

if p.len() != 0 {
let addr = { self.fields.read().await.clone() };
let additional = { self.additional.read().await.clone() };

let batch = create_record_batch(&p, &addr, &additional).unwrap();
points.push(batch);
}

let dirs = self.get_dir_by_time(start, end).unwrap_or_default();

info!("dirs: {:?} {:?} {:?}", start, end, dirs);

for s in dirs {
let mut files = list_files_in_directory(&s).unwrap();
files.reverse();

let mut v = vec![];

for s in files {
if s.extension().unwrap() == "parquet" {
let f = self
.files
.optionally_get_with(format!("{:?}", s), async {
let f = add_file(Path::new(&s)).unwrap();
Some((String::from(f.0), f.1))
})
.await
.unwrap();

v.push(PartitionedFile::new(f.0.clone(), f.1));
}
}
file_scan_config = file_scan_config.with_file_group(v);
}

break;
}
}

let exec: Arc = ParquetExec::builder(file_scan_config)
.with_predicate(predicate)
.build_arc();

// count error

let memory =
MemoryExec::try_new(&vec![points], self.schema(), projection.cloned()).unwrap();

let union = Arc::new(UnionExec::new(vec![Arc::new(memory), exec]));

Ok(union)
}
```

### Describe the solution you'd like

_No response_

### Describe alternatives you've considered

_No response_

### Additional context

_No response_

The issue provides a Rust scan implementation using UnionExec, MemoryExec, ParquetExec, and mentions SortPreservingMergeExec, but names no repository file or test. Start by locating examples and tests for these execution plans, then determine how a multiple-source example should express deduplication across fields. Done means an agreed example exists and demonstrates both source merging and field-based deduplication.

Written by the indexing model from the issue text.