Repository navigation
Conversation
… before planning them A query request names the number of partitions of each stage and exchange, the batch size, and the operators of each stage's block. The scheduler allocated for each partition and planned each block as given, so a count of -1 became a usize::MAX allocation, 2^31 partitions exhausted memory, and a block with fewer input leaves than its stage declares failed only when it ran. Bound the partition count of a stage and of a hash exchange to between 1 and 65536, bound spark.comet.batchSize to between 1 and 1048576, require a block and each of its partition plans to read as many inputs as the stage declares, and refuse a block that writes files (a Parquet writer, an Iceberg write or a shuffle writer).
Limit the size of a query the Flight service accepts with COMET_SCHEDULER_FLIGHT_MAX_REQUEST_BYTES (512 MiB by default, where there was no limit), and add COMET_SCHEDULER_QUERY_TIMEOUT_SECONDS to fail and cancel a query that runs too long. The scheduler also cancels a job when the driver that submitted it disconnects, instead of letting it run on. State in the README and the binaries' documentation that the cluster has no authentication or TLS and must run on a trusted network, and document the new settings and the refusals.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #7288.
Rationale for this change
The Comet Native Executors scheduler and executors accept a query request from the network. The counts, sizes and operators in it must be bounded and checked before the planner allocates for them or runs them. Today they are not:
num_partitionsof-1becameusize::MAXpartitions, and the scheduler tried to allocate for each one.2^31partitions, or aspark.comet.batchSizeof0orusize::MAX, was accepted and sized every buffer from it.stage.inputswas accepted, although the block is fed from the stage's inputs in order.usize::MAX), and a query ran for as long as it took, even after the driver that submitted it disconnected.What changes are included in this PR?
native/core/src/execution/native_executors:MAX_PARTITIONS(65536) and aMAX_BATCH_SIZE(1048576).partition_descriptorrefuses a stage with no partitions or more thanMAX_PARTITIONS, andfragment_sessionrefuses aspark.comet.batchSizeoutside1..=MAX_BATCH_SIZE.native/native-executors/src/assembly.rs: validates a hash exchange'snum_partitions(negative, zero and over the cap) and a stage's partition count. A newcheck_blockrequires a stage's block, and each entry ofpartition_plans, to read as many inputs as the stage declares, and refuses aParquetWriter,IcebergWriteorShuffleWriteroperator.comet-scheduler:COMET_SCHEDULER_FLIGHT_MAX_REQUEST_BYTESlimits the size of a request (512 MiB by default, replacing no limit).COMET_SCHEDULER_QUERY_TIMEOUT_SECONDSfails and cancels a query that runs longer (default 0: no limit). A query is also cancelled when the driver that submitted it disconnects.native/native-executors/README.mdand the binaries' documentation: the cluster has no authentication or TLS and must run on a trusted network, the new settings, the refusals above, and the wording on scalar subqueries.How are these changes tested?
Run locally:
cargo fmt --check,cargo clippy --all-targets -- -D warningsandcargo check --all-targetsinnative/native-executors.cargo test --test assembly --libinnative/native-executors: 20 tests intests/assembly.rsand 2 unit tests of the query timeout insrc/frontend.rs. The new tests cover bounded hash exchange and stage partition counts (includingi32::MIN,-1,0, one over the cap andu32::MAX), the batch size bound, a block and a partition plan that read the wrong number of inputs, and the three refused writers at the root of a block, below another operator and in a partition's own plan.cargo clippy -p datafusion-comet --all-targets -- -D warningsandcargo test -p datafusion-comet --lib native_executorsinnative.prettier --checkon the README.Not run:
tests/cluster.rs, which starts the scheduler and executor binaries and needs sockets, which the environment I developed in does not allow. It compiles (cargo check --all-targets). The newrefuses_a_query_larger_than_the_request_limittest in it, and the cancellation of a job when a driver disconnects or a query times out, which no test exercises, have not been run.This is split out of #7289 so that it can be reviewed on its own against the published Ballista 55.0.0 crates, which this workspace builds on. #7289 carries the limits for pipelined mode.