You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@sedona.apache.org by "zhangfengcdt (via GitHub)" <gi...@apache.org> on 2025/11/24 22:56:17 UTC
[PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
zhangfengcdt opened a new pull request, #362:
URL: https://github.com/apache/sedona-db/pull/362
This PR introduces an optional sequential index building mode for spatial joins as an alternative to the default parallel implementation.
A new use_sequential_index_build boolean flag (default: false) controls the index building strategy. The sequential implementation via build_index_sync collects partitions one-by-one without spawning async tasks, supporting execution contexts that lack full async runtime support.
Key types and functions are now publicly exported: build_index, build_index_sync, SpatialIndex, SpatialIndexBuilder, SpatialJoinBuildMetrics, and SpatialPredicate, enabling external components to programmatically build and manage spatial indexes.
This PR is to support external use of the spatial join executor in other query engine projects (single or distributed), such as Apache Datafusion Comet.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2602033344
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,14 +33,68 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Sequential version of build_index that doesn't spawn tasks.
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_seq(
context: Arc<TaskContext>,
build_schema: SchemaRef,
build_streams: Vec<SendableRecordBatchStream>,
spatial_predicate: SpatialPredicate,
join_type: JoinType,
probe_threads_count: usize,
metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ false, // concurrent = false
+ )
+ .await
+}
+
+/// Concurrent version of build_index that spawns tasks for parallel collection.
+pub async fn build_index(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ true, // concurrent = true
+ )
+ .await
+}
+
+/// Internal implementation of build_index with configurable concurrency.
+///
+/// # Arguments
+/// * `concurrent` - If true, uses `collect_all` which spawns tasks for parallel collection.
+/// If false, collects partitions sequentially (for JNI/embedded contexts).
+#[allow(clippy::too_many_arguments)]
+async fn build_index_impl(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+ concurrent: bool,
Review Comment:
I also prefer renaming `use_sequential_index_build ` to `concurrent_build_side_collection`, since `concurrent_build_side_collection` accurately signifies what operation is impacted by this configuration.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2579539114
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,7 +33,72 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Synchronous version of build_index that doesn't spawn tasks
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_sync(
Review Comment:
I tried to reproduce this by spawning some async tasks in the native comet scan operator, but Comet tests involving native scan operator finished successfully and did not hang. I have also tried setting env `COMET_WORKER_THREADS=1` and still could not observe hanging. The problem could be in somewhere else and not related to spawning async tasks.
```rust
impl Stream for ScanStream<'_> {
type Item = DataFusionResult<RecordBatch>;
fn poll_next(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// ...
println!("polling batches from scan: spawn an async task");
let mut join_set = JoinSet::new();
for k in 0..10 {
let res = join_set.spawn(async move {
println!("hello {}", k);
k
});
println!("spawned async task: {:?}", res);
}
let fut = join_set.join_all();
let pin_fut = pin!(fut);
let res = ready!(pin_fut.poll(ctx));
println!("async task result: {:?}", res);
// ...
}
// ...
}
```
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2558178735
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,7 +33,72 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Synchronous version of build_index that doesn't spawn tasks
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_sync(
Review Comment:
This newly added `build_index_sync` is also an async function. Why can we not directly use `build_index`, which is also an async function?
For the use case of Comet, I believe that the DataFusion physical plans constructed by Comet are all single partition. What problem have you encountered when using `build_index`?
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2601977113
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,14 +33,68 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Sequential version of build_index that doesn't spawn tasks.
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_seq(
context: Arc<TaskContext>,
build_schema: SchemaRef,
build_streams: Vec<SendableRecordBatchStream>,
spatial_predicate: SpatialPredicate,
join_type: JoinType,
probe_threads_count: usize,
metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ false, // concurrent = false
+ )
+ .await
+}
+
+/// Concurrent version of build_index that spawns tasks for parallel collection.
+pub async fn build_index(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ true, // concurrent = true
+ )
+ .await
+}
+
+/// Internal implementation of build_index with configurable concurrency.
+///
+/// # Arguments
+/// * `concurrent` - If true, uses `collect_all` which spawns tasks for parallel collection.
+/// If false, collects partitions sequentially (for JNI/embedded contexts).
+#[allow(clippy::too_many_arguments)]
+async fn build_index_impl(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+ concurrent: bool,
Review Comment:
After looking closer to the code, I found that we actually don't need to define a new `build_index_impl` and `build_index_seq`, and introduce all the replicated code in `fn execute` and `fn execute_knn`.
We have access to the spatial join config by accessing the `context`. `build_index` already has code for extracting `sedona_options` from it. We can collect the build side partitions sequentially or collect them concurrently by spawning a number of tasks according to the `use_sequential_index_build` configuration value.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2607815351
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,14 +33,68 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Sequential version of build_index that doesn't spawn tasks.
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_seq(
context: Arc<TaskContext>,
build_schema: SchemaRef,
build_streams: Vec<SendableRecordBatchStream>,
spatial_predicate: SpatialPredicate,
join_type: JoinType,
probe_threads_count: usize,
metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ false, // concurrent = false
+ )
+ .await
+}
+
+/// Concurrent version of build_index that spawns tasks for parallel collection.
+pub async fn build_index(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+) -> Result<SpatialIndex> {
+ build_index_impl(
+ context,
+ build_schema,
+ build_streams,
+ spatial_predicate,
+ join_type,
+ probe_threads_count,
+ metrics,
+ true, // concurrent = true
+ )
+ .await
+}
+
+/// Internal implementation of build_index with configurable concurrency.
+///
+/// # Arguments
+/// * `concurrent` - If true, uses `collect_all` which spawns tasks for parallel collection.
+/// If false, collects partitions sequentially (for JNI/embedded contexts).
+#[allow(clippy::too_many_arguments)]
+async fn build_index_impl(
+ context: Arc<TaskContext>,
+ build_schema: SchemaRef,
+ build_streams: Vec<SendableRecordBatchStream>,
+ spatial_predicate: SpatialPredicate,
+ join_type: JoinType,
+ probe_threads_count: usize,
+ metrics: ExecutionPlanMetricsSet,
+ concurrent: bool,
Review Comment:
Cool! I've changed the code to get concurrent_build_side_collection directly in the build_index function and it looks much cleaner now.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2578237245
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,7 +33,72 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Synchronous version of build_index that doesn't spawn tasks
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_sync(
Review Comment:
@Kontinuation Yes, I think the function name might be a bit confusing, I can rename it to `build_index_sequential` to better convey that it avoids task spawning.
So basically, the reason we need this (instead of using build_index directly) is because the limitation with JoinSet::spawn() in JNI contexts: when running in a Spark/Comet JNI context, JoinSet::spawn() fails because:
1. The tokio runtime in JNI contexts may be single-threaded or have limited threading capabilities
2. JoinSet::spawn() requires a multi-threaded runtime to spawn new tasks
Even with single-partition DataFusion plans (which is typical for Comet), collect_all still calls JoinSet::spawn():
```
// In collect_all:
let mut join_set = JoinSet::new();
for ... {
join_set.spawn(async move { ... }); // <- This fails in JNI contexts
}
join_set.join_all().await; // Wait for all tasks
```
The issue isn't about parallelism benefits (single partition means nothing to parallelize anyway), but about JoinSet::spawn() itself not working in the JNI execution environment.
When I call `build_index` directly in JNI context, it hanged forever. I guess this might be caused by a deadlock by JoinSet::spawn().
In a JNI context:
1. When JoinSet::spawn() is called, it schedules new tasks to run on the runtime's worker threads
2. Then join_all().await blocks the current thread waiting for those spawned tasks
3. But if the current thread IS the only worker thread (or all workers are blocked), no one can execute the spawned tasks
4. Deadlock: the current thread waits for tasks that need the current thread to execute
It would be ideal to reuse build_index if there is a workaround for it, but I also think using a clean `build_index_sequential` makes sense because this is called in a distributed framework and the parallelism is provided and dictated by the distributed schedulings themselves.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on PR #362:
URL: https://github.com/apache/sedona-db/pull/362#issuecomment-3613839424
> `build_index_sync` and `build_index` has lots of duplicated code. Can we add a parameter to `build_index` to switch between the concurrent mode and sequential mode?
@Kontinuation I have refactored the build_index_sync to (1) rename it to build_index_seq (2) add a new flag to control concurrency (3) remove duplicated logic.
Could you please take a look again? Thanks!
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt merged PR #362:
URL: https://github.com/apache/sedona-db/pull/362
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2608835924
##########
rust/sedona-common/src/option.rs:
##########
@@ -64,6 +64,10 @@ config_namespace! {
/// The execution mode determining how prepared geometries are used
pub execution_mode: ExecutionMode, default = ExecutionMode::Speculative(DEFAULT_SPECULATIVE_THRESHOLD)
+ /// Collect build side partitions concurrently (using spawned tasks).
+ /// Set to false for JNI/embedded contexts without async runtime support.
+ pub concurrent_build_side_collection: bool, default = true
Review Comment:
As I mentioned in another PR, concurrent build side collection does not work for Comet not because of it does not have async runtime support. The `build_index` function is async so we always need async runtime support.
##########
rust/sedona-spatial-join/src/exec.rs:
##########
@@ -107,7 +107,7 @@ fn determine_knn_build_probe_plans<'a>(
/// 2. **Probe Phase**: Each geometry from the right table is used to query the spatial index
/// 3. **Refinement**: Candidate pairs from the index are refined using exact spatial predicates
/// 4. **Output**: Matching pairs are combined according to the specified join type
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct SpatialJoinExec {
Review Comment:
Making an ExecutionPlan `Clone` is quite unusual. The other execution plans in DataFusion are not `Clone` and they still integrate well with other systems such as Comet. Is there any specific reasons for doing this?
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2579195636
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,7 +33,72 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Synchronous version of build_index that doesn't spawn tasks
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_sync(
Review Comment:
Comet is using a multi-thread Tokio runtime according to [jni_api.rs](https://github.com/apache/datafusion-comet/blob/0.12.0/native/core/src/execution/jni_api.rs#L90-L102), and the Tokio runtime will use default config if `COMET_*` environment variables are not present.
The problem may not be the usage of single-threaded runtime, probably there are too-many in-flight queries and the workers pool have been exhausted. No matter what the actual problem is, a sequential index building option is always a good thing to have.
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(rust/sedona-spatial-join): Support sequential spatial index building option [sedona-db]
Posted by "zhangfengcdt (via GitHub)" <gi...@apache.org>.
zhangfengcdt commented on code in PR #362:
URL: https://github.com/apache/sedona-db/pull/362#discussion_r2585886974
##########
rust/sedona-spatial-join/src/build_index.rs:
##########
@@ -33,7 +33,72 @@ use crate::{
spatial_predicate::SpatialPredicate,
};
-pub(crate) async fn build_index(
+/// Synchronous version of build_index that doesn't spawn tasks
+/// Used in execution contexts without async runtime support (e.g., Spark/Comet JNI)
+pub async fn build_index_sync(
Review Comment:
I see your tests directly places JoinSet::spawn directly inside poll_next. The hang I observed might not be directly caused by JoinSet::spawn itself. The difference might be:
- Your test: Spawns directly inside poll_next using the provided Context
- Spatial join: Uses OnceAsync for lazy initialization, which has its own coordination logic
The interaction between block_on(async { poll!(...) }), OnceAsync, and JoinSet might behave differently than direct spawning in poll_next.
But in SpatialJoinExec, build_index is called through OnceAsync, which has its own coordination logic. The execution path is:
1. JNI calls get_runtime().block_on(async { poll!(stream.next()) })
2. Stream's poll_next triggers OnceAsync
3. OnceAsync calls build_index → collect_all → JoinSet::spawn
--
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: issues-unsubscribe@sedona.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org