dongjoon-hyun commented on code in PR #59251:
URL: https://github.com/apache/spark/pull/59251#discussion_r4209638526
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
+ * a transform Spark cannot evaluate. Keeping it would only add comparisons
nobody uses. A sort
+ * order that Spark derives from a transform key is even constant within
each partition. Revisit
+ * this if an ordering over a transform becomes a real requirement.
*/
override def outputOrdering: Seq[SortOrder] = {
+ def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
(ordering, outputPartitioning) match {
case (Some(o), p) =>
- val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
+ val (prefix, rest) =
+ o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
Review Comment:
Minor: a sort order on a transform that is itself a partition key ends the
leading run here, although it is constant within each partition, which is the
reason L142-143 give for keeping the sort orders on a partition key. So the
sort orders after it that still hold are dropped. With keys `[days(ts)]` and a
reported `[days(ts), id]`, the scan reports `[]` instead of `[id]`. With keys
`[years(ts), id]` and a reported `[years(ts), id, name]`, it reports `[id]`
instead of `[id, name]`. A reported `[days(ts)]` over keys `[days(ts), id]`
also gives `[]`, while no report would derive `[id]`;
V2ScanPartitioningAndOrdering.scala:98-101 uses `None` for a dropped report for
that reason. This is not a regression, since none of these orderings satisfied
anything before, and it comes from the cut "before the first sort key that
holds a transform" that I suggested in item 5. Matching a transform sort order
to a transform key would need `isSameFunction` plus semantically equal children
rather tha
n `ExpressionSet`, since the two are bound separately and
`BoundFunction.equals` is only a SHOULD (BoundFunction.java:112-124).
Could we skip a sort order on a partition key instead of ending the run
there, e.g. `prefix ++ rest.filter(o => isKey(o) && usable(o)) ++
rest.takeWhile(o => isKey(o) || usable(o)).filterNot(isKey)` with `usable`
being the `span` predicate, and fall back to the derived ordering when nothing
of a non-empty report survives? A follow-up JIRA would be fine too.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -6571,6 +6571,102 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59995: a scan's output ordering drops sort orders that hold a
partition transform") {
+ // `t1` reports a transform in the middle, so the leading run of its
ordering stops there.
+ // `t2` reports a transform key first. Each split holds a single key, so
the sort order on `id`
+ // after it still holds. `t3` reports no ordering, so the scan derives one
from its keys. In
+ // each case a sort on `id` right above the scan needs no `SortExec`.
+ val table1 = "transform_order_t1"
+ val table2 = "transform_order_t2"
+ val table3 = "transform_order_t3"
+ def asc(expr: Expression): SortOrder =
+ sort(expr, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
+ createTable(table1, columns, Array(identity("id")),
+ Array(asc(FieldReference("id")), asc(years("ts")),
asc(FieldReference("data"))))
+ createTable(table2, columns, Array(years("ts"), identity("id")),
+ Array(asc(years("ts")), asc(FieldReference("id"))))
+ createTable(table3, columns, Array(days("ts"), identity("id")))
+ Seq(table1, table2, table3).foreach { table =>
+ sql(s"INSERT INTO testcat.ns.$table VALUES (1, 'aa', cast('2020-01-01'
as timestamp))")
+ val df = sql(s"SELECT id, data, ts FROM testcat.ns.$table")
+ val scan = collectScans(df.queryExecution.executedPlan).head
+ assert(scan.outputOrdering.map(_.child.sql) === Seq("id"), table)
+ val sorted = df.sortWithinPartitions("id").queryExecution.executedPlan
+ assert(collect(sorted) { case s: SortExec => s }.isEmpty, table)
+ }
+ }
+
+ /**
+ * Inserts three items, two of them with `id` 1 on their own splits, so
grouping the splits by
+ * `id` coalesces them. Then joins `items` with `purchases` on
`joinCondition` and checks that
+ * the plan k-way merges over `expectedOrdering`, the columns the scan's
ordering keeps before
+ * its transform.
+ */
+ private def checkKWayMergeBeforeTransform(
+ joinCondition: String,
+ expectedOrdering: Seq[String]): Unit = {
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 10.0, cast('2021-01-01' as timestamp)), " +
+ "(1, 'ab', 11.0, cast('2022-01-01' as timestamp)), " +
+ "(2, 'bb', 20.0, cast('2021-01-01' as timestamp))")
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("i", "p")}
+ |i.id, i.name, i.arrive_time
+ |FROM testcat.ns.$items i JOIN testcat.ns.$purchases p ON
$joinCondition
+ |""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(1, "aa", Timestamp.valueOf("2021-01-01 00:00:00")),
+ Row(1, "ab", Timestamp.valueOf("2022-01-01 00:00:00")),
+ Row(2, "bb", Timestamp.valueOf("2021-01-01 00:00:00"))))
+ val merging = collectAllGroupPartitions(df.queryExecution.executedPlan)
+ .filter(_.enableSortedMerge)
+ assert(merging.length == 1, "expected one k-way merge")
+ assert(merging.head.child.outputOrdering.map(_.child.sql) ==
expectedOrdering)
+ assert(merging.head.execute().isInstanceOf[SortedMergeCoalescedRDD[_]])
+ }
+
+ test("SPARK-59995: k-way merge over a reported ordering with a partition
transform") {
+ // The join on (id, name) needs the merge, since the key ordering on id
alone is not enough.
+ // The scan reports [id, name, years(arrive_time)] and keeps [id, name].
+ val itemOrdering = Array(
+ sort(FieldReference("id"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST),
+ sort(FieldReference("name"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST),
+ sort(years("arrive_time"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
+ createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
+ val namedPurchasesColumns = Array(
+ Column.create("item_id", LongType),
+ Column.create("name", StringType))
+ createTable(purchases, namedPurchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES (1, 'aa'), (1, 'ab'), (2,
'bb')")
+
+ withSQLConf(
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key ->
"true") {
+ checkKWayMergeBeforeTransform("p.item_id = i.id AND p.name = i.name",
Seq("id", "name"))
+ }
+ }
+
+ test("SPARK-59995: k-way merge over an ordering derived from a partition
transform key") {
+ // The scan reports no ordering, so it derives [id, days(arrive_time)]
from its keys and keeps
+ // [id]. The merge must not call the transform's function. Spark cannot
call this `days` at
+ // all, since `DaysFunction` implements neither `invoke` nor
`produceResult`. The join on id
+ // projects the keys to id. With preserveKeyOrderingOnCoalesce off, the
coalesced partitions
+ // keep no ordering on id unless they are merged.
+ createTable(items, itemsColumns, Array(identity("id"),
days("arrive_time")))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 10.0, cast('2021-01-01' as timestamp)), " +
+ "(2, 20.0, cast('2021-01-01' as timestamp))")
+
+ withSQLConf(
Review Comment:
**These tests pass only where
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` defaults to true,
which is master and branch-4.x since SPARK-59396, so they fail if this fix is
backported to branch-4.3 or branch-4.2.** Both branches default the conf to
false. SPARK-58324, SPARK-59279, SPARK-59899 and SPARK-59905 all went to both
branches, and this suite applies cleanly to them. With the conf off, `t3`
(L6588) reports no ordering, so `outputOrdering` is empty and L6593 fails. Here
the scan derives no ordering, so `kWayMergeIsFeasible` is false, a `SortExec`
replaces the merge, and `assert(merging.length == 1, ...)` at L6624 fails. The
SPARK-56241 tests in this suite set the conf explicitly (L6342, L6388). On
branch-4.2, this test also needs
`V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS`, but that one fails at
compile time.
Could we wrap the loop at L6589-6596 in
`withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key ->
"true")` and add the same entry here?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
+ * a transform Spark cannot evaluate. Keeping it would only add comparisons
nobody uses. A sort
+ * order that Spark derives from a transform key is even constant within
each partition. Revisit
+ * this if an ordering over a transform becomes a real requirement.
*/
override def outputOrdering: Seq[SortOrder] = {
+ def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
(ordering, outputPartitioning) match {
case (Some(o), p) =>
- val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
+ val (prefix, rest) =
+ o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
p match {
case k: KeyedPartitioning if rest.nonEmpty =>
- val keyExprs = ExpressionSet(k.expressions)
+ val keyExprs =
ExpressionSet(k.expressions.filterNot(holdsTransform))
Review Comment:
Nit: BoundFunction.java:121 still lists "retaining a reported ordering that
matches the partitioning" among the benefits of a semantic `equals`. That
refers to this `keyExprs.contains` match and the one at
GroupPartitionsExec.scala:397-398, but `keyExprs` now holds no transform, so
whether a reported ordering is retained no longer depends on `equals`.
Could we drop that bullet, unless item 15 brings back an `equals`-based
match?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -6571,6 +6571,102 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59995: a scan's output ordering drops sort orders that hold a
partition transform") {
+ // `t1` reports a transform in the middle, so the leading run of its
ordering stops there.
+ // `t2` reports a transform key first. Each split holds a single key, so
the sort order on `id`
+ // after it still holds. `t3` reports no ordering, so the scan derives one
from its keys. In
+ // each case a sort on `id` right above the scan needs no `SortExec`.
+ val table1 = "transform_order_t1"
+ val table2 = "transform_order_t2"
+ val table3 = "transform_order_t3"
+ def asc(expr: Expression): SortOrder =
+ sort(expr, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
+ createTable(table1, columns, Array(identity("id")),
+ Array(asc(FieldReference("id")), asc(years("ts")),
asc(FieldReference("data"))))
+ createTable(table2, columns, Array(years("ts"), identity("id")),
+ Array(asc(years("ts")), asc(FieldReference("id"))))
+ createTable(table3, columns, Array(days("ts"), identity("id")))
+ Seq(table1, table2, table3).foreach { table =>
+ sql(s"INSERT INTO testcat.ns.$table VALUES (1, 'aa', cast('2020-01-01'
as timestamp))")
+ val df = sql(s"SELECT id, data, ts FROM testcat.ns.$table")
Review Comment:
Nit: these tests keep the transform's column in the scan output only through
their SELECT lists, here and `i.arrive_time` at L6615. With it pruned, the
pruned-column handling from SPARK-59899 already gives the same orderings on
master: `[id]` for all three tables here, and `[id, name]` and `[id]` in the
merge tests. So a later cleanup of a SELECT list would silently stop these
tests from guarding the fix, and this one has no `checkAnswer` to notice. The
SPARK-59899 test asserts the opposite precondition (L3910-3911).
Could we assert it here too, e.g. `assert(scan.output.exists(_.name ==
"ts"), s"test setup: ts stays in the output of $table")`, and mention it in the
scaladoc of `checkKWayMergeBeforeTransform`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
+ * a transform Spark cannot evaluate. Keeping it would only add comparisons
nobody uses. A sort
+ * order that Spark derives from a transform key is even constant within
each partition. Revisit
+ * this if an ordering over a transform becomes a real requirement.
*/
override def outputOrdering: Seq[SortOrder] = {
+ def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
(ordering, outputPartitioning) match {
case (Some(o), p) =>
- val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
+ val (prefix, rest) =
+ o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
p match {
case k: KeyedPartitioning if rest.nonEmpty =>
- val keyExprs = ExpressionSet(k.expressions)
+ val keyExprs =
ExpressionSet(k.expressions.filterNot(holdsTransform))
prefix ++ rest.filter(order => keyExprs.contains(order.child))
case _ => prefix
}
case (_, k: KeyedPartitioning) if
conf.v2BucketingPartitionKeyOrderingEnabled =>
Review Comment:
Minor: the PR description does not say since when the failure exists or
where the ordering change is on by default. The k-way merge (SPARK-55715) and
the key-derived ordering (SPARK-56241) are both in 4.2.0, so the failure exists
since 4.2.0 with
`spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` on, while
this conf defaults to true only on master and branch-4.x. The new `[id]` can
also turn a hash aggregate into a sort aggregate.
`spark.sql.execution.replaceHashWithSortAgg` defaults to true on master,
branch-4.x and branch-4.3, and ReplaceHashWithSortAgg.scala:58-65 switches when
the child's ordering satisfies the grouping, so `SELECT id, max(ts) FROM t
GROUP BY id` over keys `[years(ts), id]` now plans a `SortAggregateExec`. The
template asks whether a change is user-facing compared to the released versions.
Could we add both to the user-facing section?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
+ * a transform Spark cannot evaluate. Keeping it would only add comparisons
nobody uses. A sort
Review Comment:
Nit: "Keeping it would only add comparisons nobody uses." repeats the
argument of L150 two sentences later, and "A sort order that Spark derives from
a transform key is even constant within each partition." repeats L144-146:
every derived sort order is constant within a partition, identity keys
included, so "even" reads as a reason specific to transforms, which it is not.
"only" also understates it: a kept transform ahead of a key hides that key from
prefix matching, as `[years(ts), id]` vs `[id]` shows.
Could L150-154 be shortened to something like "No operator requires an
ordering over a transform (the write path sorts by the transform's function
call when Spark can call it), so keeping it would add comparisons nobody uses,
and dropping it also handles a transform Spark cannot evaluate. Revisit this if
such a requirement appears."?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
Review Comment:
Minor: a correction to my item 5, which said that "the write path replaces
transforms with their calls". That holds only when Spark can call the function,
and so does "The write path sorts by the transform's function call instead"
here. `resolvedFunction` is defined only for a `ScalarFunction`
(TransformExpression.scala:177-182), and `resolveTransformExpression`
(DistributionAndOrderingUtils.scala:97-99) keeps any other transform in the
write's local sort. Take a function `f` that binds to a non-scalar
`BoundFunction` with a semantic `equals`, a table `t1` that reports the
ordering `[f(id)]`, and a table `t2` that requires `[f(id)]` with an
unspecified distribution. Before this PR, `INSERT INTO t2 SELECT * FROM t1`
worked, because `RemoveRedundantSorts` (RemoveRedundantSorts.scala:52-55)
dropped the write's `SortExec` against the scan's ordering. Now the scan drops
that sort order, so the `SortExec` stays and fails to evaluate `f`. No
connector I know binds a transform to a non-sc
alar function, and such a write already failed from any other source, so I
think only the wording needs to change.
Could L150-151 say "The write path sorts by the transform's function call
instead, when Spark can call the function"?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
+ * In a reported ordering it also ends the leading run, like a sort order
over a pruned column.
+ * This loses nothing today, since no operator requires an ordering over a
transform. The write
+ * path sorts by the transform's function call instead. Dropping it is also
a safe way to handle
+ * a transform Spark cannot evaluate. Keeping it would only add comparisons
nobody uses. A sort
+ * order that Spark derives from a transform key is even constant within
each partition. Revisit
+ * this if an ordering over a transform becomes a real requirement.
*/
override def outputOrdering: Seq[SortOrder] = {
+ def holdsTransform(e: Expression): Boolean =
e.exists(_.isInstanceOf[TransformExpression])
(ordering, outputPartitioning) match {
case (Some(o), p) =>
- val (prefix, rest) = o.span(_.references.subsetOf(outputSet))
+ val (prefix, rest) =
+ o.span(order => order.references.subsetOf(outputSet) &&
!holdsTransform(order.child))
p match {
case k: KeyedPartitioning if rest.nonEmpty =>
- val keyExprs = ExpressionSet(k.expressions)
+ val keyExprs =
ExpressionSet(k.expressions.filterNot(holdsTransform))
prefix ++ rest.filter(order => keyExprs.contains(order.child))
case _ => prefix
}
case (_, k: KeyedPartitioning) if
conf.v2BucketingPartitionKeyOrderingEnabled =>
- k.expressions.map(SortOrder(_, Ascending))
+ k.expressions.filterNot(holdsTransform).map(SortOrder(_, Ascending))
Review Comment:
Minor: with transform keys left out here, some docs now overstate what the
derived ordering gives. The 4.4 migration note (docs/sql-migration-guide.md:45)
says the scan "now also reports itself sorted by its partition key
expressions", and the `partitionKeyOrdering` doc (SQLConf.scala:2591-2595) and
docs/sql-performance-tuning.md:711 say the same. A table partitioned only by
transforms, such as `days(ts)` or `bucket(8, id)`, now derives no ordering, so
the conf does nothing for it, and keys `[years(ts), id]` derive `[id]`.
SupportsReportOrdering.java:36-42 could also tell connector authors that a sort
order over a partition transform is not used.
Could we say in those docs that partition transforms are left out, e.g.
"sorted by those of its partition key expressions that are columns (a partition
transform such as `days(ts)` or `bucket(8, id)` is left out)"?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -144,19 +144,29 @@ trait DataSourceV2ScanExecBase
* is a `KeyedPartitioning` and
`spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled`
* is on, each partition contains rows where the key expressions evaluate to
a single constant
* value, so the data is trivially sorted by those expressions within the
partition.
+ *
+ * Either way, a sort order that holds a partition transform is dropped,
even on a partition key.
Review Comment:
Nit: this makes the coalescing-branch comment at
GroupPartitionsExec.scala:391-395 stale. It says the child's outputOrdering
"should already be in sync with the partitioning (either reported by the source
or derived from it in DataSourceV2ScanExecBase)", but this scan now leaves out
every transform key, so the transform entries of `keyExprs` there never match.
For example, keys `[years(ts)]` with `preserveKeyOrderingOnCoalesce` on
reported `[years(ts)]` before and report `[]` now. That file is already in this
PR.
Could that comment say the child's ordering is in sync with the
partitioning's column keys, since this scan drops every sort order over a
partition transform?
--
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]
