peter-toth commented on code in PR #59789:
URL: https://github.com/apache/spark/pull/59789#discussion_r4378624128
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11064,6 +11064,69 @@ class KeyGroupedPartitioningSuite
s"the aggregate's side is grouped over the join:\n$plan")
}
}
+
+ test("SPARK-60260: SPJ keeps right results when v2 bucketing turns off after
planning") {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+ withTable(s"testcat.ns.$items", s"testcat.ns.$purchases") {
+ // Seven keys, one split each on both sides, so the join needs no
GroupPartitionsExec.
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'a', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 'b', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'c', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'd', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 'e', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 'f', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 'g', 10.0, cast('2020-01-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 1.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 2.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 3.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 4.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 5.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 6.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 7.0, cast('2020-01-01' as timestamp))")
+
+ // Plans the join with v2 bucketing on, asserts it is the shape the
bug needs -- a
+ // storage-partitioned join (sort-merge join, no shuffle, no
GroupPartitionsExec) with
+ // dynamic partition pruning on the keyed scan -- then executes it
with bucketing turned off
+ // and checks the rows. Asserting the plan keeps the test from passing
without actually
+ // exercising the keyed-scan rebuild path.
+ def checkWithBucketingTurnedOff(filter: String, expected: Seq[Row]):
Unit = {
+ val df = sql(
+ s"SELECT i.id, i.name, p.price FROM testcat.ns.$items i " +
+ s"JOIN testcat.ns.$purchases p ON i.id = p.item_id WHERE $filter")
+ val planned = stripAQEPlan(df.queryExecution.executedPlan)
+ assert(collect(planned) { case j: SortMergeJoinExec => j }.nonEmpty,
+ s"expected a sort-merge join:\n$planned")
+ assert(collectShuffles(planned).isEmpty,
+ s"expected a storage-partitioned join with no shuffle:\n$planned")
+ assert(collectGroupPartitions(planned).isEmpty,
+ s"expected one split per key, no GroupPartitionsExec:\n$planned")
+ assert(collectScans(planned).exists(_.runtimeFilters.nonEmpty),
+ s"expected dynamic partition pruning on the keyed scan:\n$planned")
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
+ checkAnswer(df, expected)
+ }
+ }
+
+ // All keys survive the filter: on master this returns 2 rows instead
of 7.
+ checkWithBucketingTurnedOff("p.price > 0.0",
+ Seq(Row(1L, "a", 1.0f), Row(2L, "b", 2.0f), Row(3L, "c", 3.0f),
Row(4L, "d", 4.0f),
+ Row(5L, "e", 5.0f), Row(6L, "f", 6.0f), Row(7L, "g", 7.0f)))
+ // A filter that prunes splits: on master this fails with "Can't zip
RDDs with unequal
+ // numbers of partitions"; only item ids 6 and 7 survive.
+ checkWithBucketingTurnedOff("p.price > 5.0", Seq(Row(6L, "f", 6.0f),
Row(7L, "g", 7.0f)))
Review Comment:
**Finding 2.** The fix covers two more shapes, and the test pins neither.
Both ran like `checkWithBucketingTurnedOff`: planned with bucketing on, then
collected with it off, AQE on.
| Shape | master | here |
|---|---|---|
| `items` holds two rows per id, so a `GroupPartitionsExec` sits above the
pruned scan | `[INTERNAL_ERROR] GroupPartitionsExec's child no longer reports
the partitioning it was planned over`, for both filters | right (14 rows, 4
rows) |
| `purchases` is not partitioned and
`spark.sql.sources.v2.bucketing.shuffle.enabled` is on, so it is shuffled onto
the keyed `items` scan | 2 rows instead of 7, and `Can't zip RDDs with unequal
numbers of partitions: List(2, 7)` for `p.price > 5.0` | right |
Turning the helper's `collectGroupPartitions(planned).isEmpty` and
`collectShuffles(planned).isEmpty` into expected counts lets one helper run all
three table setups. The first one is worth pinning in particular, since it
shows the fix turns the `checkChildStillMatches` internal error into right rows.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11064,6 +11064,69 @@ class KeyGroupedPartitioningSuite
s"the aggregate's side is grouped over the join:\n$plan")
}
}
+
+ test("SPARK-60260: SPJ keeps right results when v2 bucketing turns off after
planning") {
Review Comment:
**Finding 3.** The suite keeps its runtime-filtering tests under SPJ in
`KeyGroupedPartitioningRuntimeFilterTests`
(`KeyGroupedPartitioningSuite.scala:129`), next to "SPARK-45652: SPJ should
handle empty partition after dynamic filtering". That trait runs each test
twice:
- in `KeyGroupedPartitioningRuntimeFilterSuite`, against the default catalog;
- in `KeyGroupedPartitioningCatalystRuntimeFilterSuite`, against
`InMemoryCatalystRuntimeFilterCatalog`, whose scans take Catalyst runtime
filters.
Here the test runs only against the default catalog. I ran its shape in both
suites. Against the Catalyst-filter catalog, master also returns 2 rows instead
of 7 and fails with `Can't zip RDDs`, through the
`SupportsRuntimeCatalystFiltering` path of `PushDownUtils.pushRuntimeFilters`,
and both are right here. The helpers the test uses are on
`KeyGroupedPartitioningSuiteBase`, so it should move as is.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11064,6 +11064,69 @@ class KeyGroupedPartitioningSuite
s"the aggregate's side is grouped over the join:\n$plan")
}
}
+
+ test("SPARK-60260: SPJ keeps right results when v2 bucketing turns off after
planning") {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+ withTable(s"testcat.ns.$items", s"testcat.ns.$purchases") {
+ // Seven keys, one split each on both sides, so the join needs no
GroupPartitionsExec.
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'a', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 'b', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'c', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'd', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 'e', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 'f', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 'g', 10.0, cast('2020-01-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 1.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 2.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 3.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 4.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 5.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 6.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 7.0, cast('2020-01-01' as timestamp))")
+
+ // Plans the join with v2 bucketing on, asserts it is the shape the
bug needs -- a
+ // storage-partitioned join (sort-merge join, no shuffle, no
GroupPartitionsExec) with
+ // dynamic partition pruning on the keyed scan -- then executes it
with bucketing turned off
+ // and checks the rows. Asserting the plan keeps the test from passing
without actually
+ // exercising the keyed-scan rebuild path.
+ def checkWithBucketingTurnedOff(filter: String, expected: Seq[Row]):
Unit = {
+ val df = sql(
+ s"SELECT i.id, i.name, p.price FROM testcat.ns.$items i " +
+ s"JOIN testcat.ns.$purchases p ON i.id = p.item_id WHERE $filter")
+ val planned = stripAQEPlan(df.queryExecution.executedPlan)
+ assert(collect(planned) { case j: SortMergeJoinExec => j }.nonEmpty,
+ s"expected a sort-merge join:\n$planned")
+ assert(collectShuffles(planned).isEmpty,
+ s"expected a storage-partitioned join with no shuffle:\n$planned")
+ assert(collectGroupPartitions(planned).isEmpty,
+ s"expected one split per key, no GroupPartitionsExec:\n$planned")
+ assert(collectScans(planned).exists(_.runtimeFilters.nonEmpty),
+ s"expected dynamic partition pruning on the keyed scan:\n$planned")
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
+ checkAnswer(df, expected)
+ }
Review Comment:
**Finding 4.** Two departures from the suite's other flip-after-planning
tests, "SPARK-59279: a planned k-way merge survives a later config change" and
"SPARK-60044: a planned k-way merge keeps its ordering after a later config
change":
- `checkAnswer(df, expected)` uses the default `checkToRDD = true`. That
first runs `df.materializedRdd.count()`, a new `QueryExecution` planned with
bucketing off, and only then collects the committed plan. The siblings pass
`checkToRDD = false`, so only the committed plan runs.
- The plan assertions look only at the initial AQE plan, so nothing checks
the node this PR changes, the scan `PlanAdaptiveDynamicPruningFilters`
rebuilds. Like the siblings, look at the executed plan again once the query has
run.
```suggestion
withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
// Without `checkToRDD`, so only the committed plan runs, not a
newly planned one.
QueryTest.checkAnswer(df, expected, checkToRDD = false)
// Under AQE the executed plan only reaches the rebuilt scans
once the query has run.
val executed = stripAQEPlan(df.queryExecution.executedPlan)
assert(collectScans(executed).forall(
_.outputPartitioning.isInstanceOf[KeyedPartitioning]),
s"expected the rebuilt scans to keep their keyed
layout:\n$executed")
}
```
With it the test passes here. On master the scan with the DPP filter reports
`UnknownPartitioning` after the run, so the assertion pins the rebuilt scan
itself, not only its rows. For the shuffled-onto shape in 2, only the keyed
scan reports `KeyedPartitioning`, so the check there would look at that scan
alone.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -117,9 +122,14 @@ case class BatchScanExec(
runtimeFilters.filterNot(_ ==
DynamicPruningExpression(Literal.TrueLiteral)),
output),
keyGroupedPartitioning = prunedKeyGroupedPartitioning.map(
- _.map(QueryPlan.normalizeExpressions(_, output))))
+ _.map(QueryPlan.normalizeExpressions(_, output))))(v2BucketingEnabled)
}
+ // `v2BucketingEnabled` is not in the case class product, so carry it
through the TreeNode
+ // transforms that rebuild this node (e.g. the runtime-filter rewrite at
stage creation), which
+ // would otherwise drop it back to the live-config default.
+ override def otherCopyArgs: Seq[AnyRef] = Boolean.box(v2BucketingEnabled) ::
Nil
Review Comment:
**Finding 7.** Two things in these lines:
- Without `otherCopyArgs`, a rebuild does not fall back to the live config.
`makeCopy` finds no constructor that fits, and the query fails with "Failed to
copy node ... wrong number of arguments: 6 expected: 7". I checked by removing
it.
- `TreeNode` declares `otherCopyArgs` as `protected`, and so do the other
nodes with a second parameter list, such as `WholeStageCodegenExec` and
`SubqueryAdaptiveBroadcastExec`.
Both are moot with 6.
```suggestion
// `v2BucketingEnabled` is not in the case class product, so pass it to
the TreeNode transforms
// that rebuild this node (e.g. the runtime-filter rewrite at stage
creation). `makeCopy` fails
// without it, since the constructor takes one more argument than the
product holds.
override protected def otherCopyArgs: Seq[AnyRef] =
Boolean.box(v2BucketingEnabled) :: Nil
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -53,6 +53,16 @@ trait DataSourceV2ScanExecBase
* `SupportsReportOrdering` */
def ordering: Option[Seq[SortOrder]]
+ /**
+ * Whether storage-partitioned join is on for this scan, captured when it
was planned. A scan
+ * rebuilt or re-executed after planning (for instance one that
+ * `PlanAdaptiveDynamicPruningFilters` recreates at stage creation) keeps
the decision it was
+ * planned with, rather than re-reading
`spark.sql.sources.v2.bucketing.enabled`, which can
+ * change between planning and execution and would drop the keyed layout the
join was planned
+ * with (SPARK-60260). A scan that freezes it overrides this; others use the
live config.
+ */
Review Comment:
**Finding 8.** "A scan rebuilt or re-executed after planning ... keeps the
decision it was planned with" holds for a copy made by a tree transform,
through `makeCopy` and `otherCopyArgs`. A scan that AQE re-plans is different.
`AdaptiveSparkPlanExec.reOptimize` calls `planner.plan`, which builds a fresh
`BatchScanExec(...)()` with the config as it is then. That is fine, since
`EnsureRequirements` then runs over the new plan, but the scaladoc should say
which rebuilds it covers. Moot with 6.
```suggestion
/**
* Whether storage-partitioned join is on for this scan, captured when it
was planned. A copy
* of a planned scan made by a tree transform (for instance the one
* `PlanAdaptiveDynamicPruningFilters` makes at stage creation) keeps the
decision it was
* planned with, rather than re-reading
`spark.sql.sources.v2.bucketing.enabled`, which can
* change between planning and execution and would drop the keyed layout
the join was planned
* with (SPARK-60260). A scan that AQE re-plans is a new scan and takes
the config then. A scan
* that freezes it overrides this; others use the live config.
*/
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -42,7 +43,11 @@ case class BatchScanExec(
runtimeFilters: Seq[Expression],
ordering: Option[Seq[SortOrder]] = None,
@transient table: Table,
- keyGroupedPartitioning: Option[Seq[Expression]] = None
+ keyGroupedPartitioning: Option[Seq[Expression]] = None)(
+ // A second parameter list so adding this field keeps `BatchScanExec`'s
extractor arity, leaving
+ // the many `case BatchScanExec(...)` matches untouched. Captured when the
scan is planned and
+ // carried through rebuilds (see `v2BucketingEnabled` and `otherCopyArgs`).
+ override val v2BucketingEnabled: Boolean = SQLConf.get.v2BucketingEnabled
Review Comment:
**Finding 6.** The PR adds a second parameter list, an `otherCopyArgs`
override, a `doCanonicalize` change and a trait hook to carry one decision the
node already has a field for. `DataSourceV2Strategy` can make the decision
where it builds the scan, and `keyGroupedPartitioning` can carry it:
```scala
// Decide storage-partitioned join here, once, so a rebuild of the
planned node does not
// read the config again (SPARK-60260).
val keyGroupedPartitioning = relation.keyGroupedPartitioning
.filter(_ => session.sessionState.conf.v2BucketingEnabled)
val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
relation.ordering, relation.relation.table, keyGroupedPartitioning)
```
And `DataSourceV2ScanExecBase.reportedKeyedPartitioning` drops the config
term:
```scala
case Some(exprs) if KeyedPartitioning.supportsExpressions(exprs) &&
```
What it gains:
- `BatchScanExec`'s constructor, `apply` and `copy` stay as they are. With
the second list, out-of-tree code that builds the node stops compiling
("missing argument list"), and code compiled against the 6-argument signatures
fails to link. MiMa does not see it, since `org.apache.spark.sql.execution` is
excluded. That matters most for the backports.
- `keyGroupedPartitioning` is a product field, so `makeCopy`, `copy`,
`doCanonicalize`, `equals` and `hashCode` already carry it, and the `()`
changes in the three test suites go away.
- Nothing samples `SQLConf.get` in a default argument behind
`BatchScanExec(...)()`.
With these two changes on master, the new test, both shapes in 2 against
both runtime-filter catalogs, and `KeyGroupedPartitioningSuite`,
`SparkPlanSuite`, `EnsureRequirementsSuite` and `GroupPartitionsExecSuite` pass.
The costs:
- `keyGroupedPartitioning = None` then means both "the source reports no
keys" and "SPJ is off for this scan". Its scaladoc in
`DataSourceV2ScanExecBase` would need a line for that.
- An out-of-tree `DataSourceV2ScanExecBase` that sets
`keyGroupedPartitioning` itself is no longer gated by the config. In the tree,
only `BatchScanExec` gets one.
If you keep the second parameter list, please at least drop the `=
SQLConf.get.v2BucketingEnabled` default, and pass the value at
`DataSourceV2Strategy.scala:204`, the only production call site. Then a future
rebuild through the constructor instead of `makeCopy` cannot silently go back
to the live config, which is this bug.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11064,6 +11064,69 @@ class KeyGroupedPartitioningSuite
s"the aggregate's side is grouped over the join:\n$plan")
}
}
+
+ test("SPARK-60260: SPJ keeps right results when v2 bucketing turns off after
planning") {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10") {
+ withTable(s"testcat.ns.$items", s"testcat.ns.$purchases") {
+ // Seven keys, one split each on both sides, so the join needs no
GroupPartitionsExec.
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'a', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 'b', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'c', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'd', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 'e', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 'f', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 'g', 10.0, cast('2020-01-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 1.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 2.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 3.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 4.0, cast('2020-01-01' as timestamp)), " +
+ "(5, 5.0, cast('2020-01-01' as timestamp)), " +
+ "(6, 6.0, cast('2020-01-01' as timestamp)), " +
+ "(7, 7.0, cast('2020-01-01' as timestamp))")
+
+ // Plans the join with v2 bucketing on, asserts it is the shape the
bug needs -- a
+ // storage-partitioned join (sort-merge join, no shuffle, no
GroupPartitionsExec) with
+ // dynamic partition pruning on the keyed scan -- then executes it
with bucketing turned off
+ // and checks the rows. Asserting the plan keeps the test from passing
without actually
+ // exercising the keyed-scan rebuild path.
+ def checkWithBucketingTurnedOff(filter: String, expected: Seq[Row]):
Unit = {
+ val df = sql(
+ s"SELECT i.id, i.name, p.price FROM testcat.ns.$items i " +
+ s"JOIN testcat.ns.$purchases p ON i.id = p.item_id WHERE $filter")
+ val planned = stripAQEPlan(df.queryExecution.executedPlan)
+ assert(collect(planned) { case j: SortMergeJoinExec => j }.nonEmpty,
+ s"expected a sort-merge join:\n$planned")
+ assert(collectShuffles(planned).isEmpty,
+ s"expected a storage-partitioned join with no shuffle:\n$planned")
+ assert(collectGroupPartitions(planned).isEmpty,
+ s"expected one split per key, no GroupPartitionsExec:\n$planned")
+ assert(collectScans(planned).exists(_.runtimeFilters.nonEmpty),
+ s"expected dynamic partition pruning on the keyed scan:\n$planned")
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "false") {
+ checkAnswer(df, expected)
+ }
+ }
+
+ // All keys survive the filter: on master this returns 2 rows instead
of 7.
+ checkWithBucketingTurnedOff("p.price > 0.0",
+ Seq(Row(1L, "a", 1.0f), Row(2L, "b", 2.0f), Row(3L, "c", 3.0f),
Row(4L, "d", 4.0f),
+ Row(5L, "e", 5.0f), Row(6L, "f", 6.0f), Row(7L, "g", 7.0f)))
+ // A filter that prunes splits: on master this fails with "Can't zip
RDDs with unequal
+ // numbers of partitions"; only item ids 6 and 7 survive.
Review Comment:
**Finding 9.** "on master" stops being true once this merges, while "without
the fix" stays true.
```suggestion
// All keys survive the filter: without the fix this returns 2 rows
instead of 7.
checkWithBucketingTurnedOff("p.price > 0.0",
Seq(Row(1L, "a", 1.0f), Row(2L, "b", 2.0f), Row(3L, "c", 3.0f),
Row(4L, "d", 4.0f),
Row(5L, "e", 5.0f), Row(6L, "f", 6.0f), Row(7L, "g", 7.0f)))
// A filter that prunes splits: without the fix this fails with
"Can't zip RDDs with
// unequal numbers of partitions"; only item ids 6 and 7 survive.
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
