Apache Ignite查询并行度的实现机制及结果差异咨询

Hey there! Let's tackle your two questions about Apache Ignite query parallelism—how it works under the hood, and why you might see differing results when enabling it.

Ignite's query parallelism is built around its distributed partitioned architecture, here's a breakdown of how it operates:

• Partition-based Parallel Scanning
  All data in Ignite is split into fixed-size partitions, which are distributed across cluster nodes. When parallelism is enabled, a single query is split into multiple subtasks, each responsible for scanning one or more partitions. These subtasks run concurrently on the nodes where the target partitions reside, eliminating the bottleneck of serial scanning from a single node.

• Configuration Controls
  You can adjust parallelism at two levels:

  • Per-query level: Use SqlQuery.setParallelism(int parallelism) to set the number of concurrent subtasks for a specific SQL query.
  • Global level: Configure IgniteConfiguration.setSqlQueryParallelism(int parallelism) to set the default parallelism for all SQL queries (the default value matches the number of available processors on the node).

• Task Scheduling & Result Merging
  After submitting a query, Ignite's query engine first parses the SQL to identify the set of partitions that need scanning. It then distributes scan tasks to the thread pools of the relevant nodes. Each node scans its local partitions, sends partial results back to the query initiator node, which finally merges all partial results into a single final result set.

If you're seeing significant differences between parallel and non-parallel query results, here are the most likely culprits and fixes:

• Unhandled Sorting/Aggregation Consistency
  Queries with ORDER BY or aggregation functions (like SUM, COUNT) can behave differently in parallel mode:

  • For sorting: Parallel scans return partial results, which the initiator node must sort globally. If your query relies on a specific order but doesn't explicitly define it with ORDER BY, you might see different row order (this isn't an error, but often mistaken for a result discrepancy).
  • For aggregations: If your query uses custom aggregations or window functions, ensure the logic is mergeable. Ignite combines partial aggregation results from each partition—if your custom function doesn't handle merging correctly, you'll get incorrect totals.

• Concurrent Data Modifications (Dirty Reads)
  When parallel subtasks scan different partitions simultaneously, ongoing data modifications from other transactions can lead to dirty reads (depending on your transaction isolation level). A serial query scans partitions one after another, so it might capture a more consistent snapshot of data. To fix this:

  • Use a higher isolation level like REPEATABLE_READ
  • Ensure no write operations are running during your query, or use distributed locks to protect the data set.

• Uneven Partition Distribution
  If your data is skewed (some partitions hold way more data than others), parallel scans might finish at different times. If data is written during the query, later-finishing partitions might include new data that earlier ones didn't. For JOIN queries, parallel join logic can also differ from serial—double-check your join conditions are correct and that data is evenly distributed across partitions.

• Non-Deterministic Query Logic
  Queries using non-deterministic functions (like RAND(), CURRENT_TIMESTAMP) will produce different results in parallel mode. Each subtask calls these functions independently, whereas a serial query calls them once. Fix this by moving non-deterministic logic outside the query, or ensuring the function is evaluated once globally before the query runs.

Example: Setting Query Parallelism #

// Set parallelism to 4 for a specific SQL query
SqlQuery<Person, Person> ageQuery = new SqlQuery<>(Person.class, "SELECT * FROM Person WHERE age > 30");
ageQuery.setParallelism(4);

// Configure global default parallelism in Ignite config
IgniteConfiguration igniteCfg = new IgniteConfiguration();
igniteCfg.setSqlQueryParallelism(8); // Use 8 concurrent subtasks by default

内容的提问来源于stack exchange，提问作者Muhammad Magdi Youssif