dongjoon-hyun commented on code in PR #55518:
URL: https://github.com/apache/spark/pull/55518#discussion_r4316536111
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupBasedRowLevelOperationScanPlanning.scala:
##########
@@ -92,6 +95,38 @@ object GroupBasedRowLevelOperationScanPlanning extends
Rule[LogicalPlan] with Pr
}
}
+ /**
+ * Returns the columns of `relation` that the projections and filters
directly above each read
+ * of the table use, as the rest of the query sees only their output. Only
whole columns are
+ * pruned, as the query is not rewritten for a pruned nested schema.
Review Comment:
**22.** Minor: copy-on-write narrow scans are pruned only at the top level,
which this Scaladoc states but the PR description and the public Javadoc do
not. `readAttrs` returns bare relation attributes, so
`PushDownUtils.pruneColumns` (L71) keeps every field of a struct that the
UPDATE references only through a nested field, while delta-based scans prune
nested fields through `V2ScanRelationPushDown`
(`V2ScanRelationPushDown.scala:1003-1016`). For `pk INT NOT NULL, salary INT, s
STRUCT<c1: INT, c2: INT>, dep STRING` and `UPDATE t SET salary = s.c1 WHERE pk
= 1`, the copy-on-write scan is pruned to `s STRUCT<c1: INT, c2: INT>` and the
delta scan to `s STRUCT<c1: INT>`. The nested `CHECK` test
(`GroupBasedColumnUpdateTableSuite.scala:639-646`) only checks that `s` is
read. This is not a regression, since the base reads every column, but on a
wide struct it is a large part of the I/O this feature is meant to avoid. Could
the PR description and the scan-builder paragraph of `SupportsC
olumnUpdates` (`SupportsColumnUpdates.java:41-44`) state that copy-on-write
scans are pruned only at the top level, and leave nested pruning to a separate
JIRA?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -129,6 +184,24 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
V2ExpressionUtils.resolveRef[AttributeReference](FieldReference(name),
plan)
}
+ protected def isIdentityAssignment(key: Attribute, value: Expression):
Boolean = {
Review Comment:
**20. An explicit `SET pk = pk` on an undeclared row-ID column makes the
distribution check reject a row-ID column, which the Javadoc allows.** This
method strips one `Alias` before `semanticEquals`, so it treats `SET pk = pk`,
which assignment alignment turns into `Alias(pk, "pk")`
(`TableOutputResolver.scala:331`), as an identity: `pk` is not reported in
`updatedColumns()` and does not have to be declared. `buildOriginalRowIdValues`
(L416-429) still uses `!key.semanticEquals(value)`, so for the same assignment
it adds `__original_row_id_pk` (`RewriteUpdateTable.scala:220`).
`newLazyRowIdProjection` (L390-399) then reads that column instead of `pk`, no
projection reads `pk`, `orderReadColumnsFirst` moves it after the last read
ordinal, and `validateColumnUpdateWrite` (`V2Writes.scala:157-166`) rejects a
write that clusters by `pk`.
With the setup of "distribution by an undeclared row ID column is allowed"
(`DeltaBasedColumnUpdateTableSuite.scala:346`: `column-update-req-attrs` =
`salary,dep` and `column-update-cluster-by` = `pk`), `UPDATE t SET salary = -1
WHERE pk = 1` succeeds, but `UPDATE t SET salary = -1, pk = pk WHERE pk = 1`
fails with `COLUMN_UPDATE_UNDECLARED_WRITE_REQUIREMENT_COLUMNS` and `[pk]`,
although `SupportsColumnUpdates.java:81-84` exempts row-ID columns for delta
operations. On the wide path the extra column is harmless, because the row
projection still reads `pk`.
Could `buildOriginalRowIdValues` use `isIdentityAssignment` as well, and
could that test get a `pk = pk` variant? Its other caller,
`RewriteMergeIntoTable.scala:351`, gets the same result, since an identity
assignment does not change the row ID.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -258,6 +354,407 @@ class InMemoryRowLevelOperationTable private (
}
}
+ // A delta-based operation that supports column-level updates: Spark sends
only the declared
+ // columns in the row projection instead of the full row schema. It declares
`pk` (the
+ // row-lookup key) and `dep` (an unconditionally declared base column) plus
whatever columns
+ // Spark reports as being assigned via
`RowLevelOperationInfo#updatedColumns()`.
+ class DeltaBasedColumnUpdateOperation(
+ command: Command,
+ updatedCols: Seq[NamedReference] = Nil,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty())
+ extends DeltaBasedOperation(command, options)
+ with SupportsColumnUpdates {
+ override def representUpdateAsDeleteAndInsert(): Boolean = false
+ override def requiredDataAttributes(): Array[NamedReference] = {
+ val base = Seq(FieldReference("pk"), FieldReference("dep"))
+ val baseNames = base.map(_.describe()).toSet
+ (base ++ updatedCols.filterNot(r =>
baseNames.contains(r.describe()))).toArray
+ }
+
+ private val clusterRef: NamedReference =
Option(properties.get(COLUMN_UPDATE_CLUSTER_BY))
+ .map(FieldReference(_))
+ .getOrElse(PARTITION_COLUMN_REF)
+
+ override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder = {
+ lastWriteInfo = info
+ // Capture info into a local val so nested writer/commit closures see a
stable schema
+ // even if a subsequent newWriteBuilder call mutates lastWriteInfo.
+ val capturedInfo = info
+ val capturedWriteSchema = if
(capturedInfo.columnUpdateSchema().isPresent) {
+ capturedInfo.columnUpdateSchema().get()
+ } else {
+ capturedInfo.schema()
+ }
+ new DeltaWriteBuilder {
+ override def build(): DeltaWrite =
+ new DeltaWrite with RequiresDistributionAndOrdering {
Review Comment:
**25.** Minor: the delta column-update fixtures ignore `no-metadata`, so no
test sends narrow rows through `DeltaWritingSparkTask`. With `no-metadata`, the
inherited `requiredMetadataAttributes()` is empty (L307-313), but this write
still requires clustering and ordering by `clusterRef`, which defaults to
`_partition` (L374-376), and the split write hard-codes `_partition` (item 18).
Every narrow UPDATE then fails with
`COLUMN_UPDATE_UNDECLARED_WRITE_REQUIREMENT_COLUMNS` and `[_partition]`,
whereas the parent returns a plain `DeltaWrite` (L326-329) and the
copy-on-write operation clusters by `dep` (L587-588). `WriteDeltaExec` uses
`DeltaWritingSparkTask` exactly when there is no metadata projection
(`WriteToDataSourceV2Exec.scala:435-440`), so that task never receives narrow
rows in a test. The round-4 comment
([r4074793050](https://github.com/apache/spark/pull/55518#discussion_r4074793050))
raised the same gap for copy-on-write, which now has no-metadata tests
(`GroupBasedColumnU
pdateTableSuite.scala:145`, `:181`). Similarly, L367 overrides
`representUpdateAsDeleteAndInsert()` to `false`, so `split-updates` is silently
ignored on these tables. Could the delta builders honor `no-metadata` like the
parent, and could a no-metadata variant of `DeltaBasedColumnUpdateTableSuite`
run, as `DeltaBasedNoMetadataUpdateTableSuite` does for full rows?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -258,6 +354,407 @@ class InMemoryRowLevelOperationTable private (
}
}
+ // A delta-based operation that supports column-level updates: Spark sends
only the declared
+ // columns in the row projection instead of the full row schema. It declares
`pk` (the
+ // row-lookup key) and `dep` (an unconditionally declared base column) plus
whatever columns
+ // Spark reports as being assigned via
`RowLevelOperationInfo#updatedColumns()`.
+ class DeltaBasedColumnUpdateOperation(
+ command: Command,
+ updatedCols: Seq[NamedReference] = Nil,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty())
+ extends DeltaBasedOperation(command, options)
+ with SupportsColumnUpdates {
+ override def representUpdateAsDeleteAndInsert(): Boolean = false
+ override def requiredDataAttributes(): Array[NamedReference] = {
+ val base = Seq(FieldReference("pk"), FieldReference("dep"))
+ val baseNames = base.map(_.describe()).toSet
+ (base ++ updatedCols.filterNot(r =>
baseNames.contains(r.describe()))).toArray
+ }
+
+ private val clusterRef: NamedReference =
Option(properties.get(COLUMN_UPDATE_CLUSTER_BY))
+ .map(FieldReference(_))
+ .getOrElse(PARTITION_COLUMN_REF)
+
+ override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder = {
+ lastWriteInfo = info
+ // Capture info into a local val so nested writer/commit closures see a
stable schema
+ // even if a subsequent newWriteBuilder call mutates lastWriteInfo.
+ val capturedInfo = info
+ val capturedWriteSchema = if
(capturedInfo.columnUpdateSchema().isPresent) {
+ capturedInfo.columnUpdateSchema().get()
+ } else {
+ capturedInfo.schema()
+ }
+ new DeltaWriteBuilder {
+ override def build(): DeltaWrite =
+ new DeltaWrite with RequiresDistributionAndOrdering {
+
+ override def requiredDistribution(): Distribution = {
+ Distributions.clustered(Array(clusterRef))
+ }
+
+ override def requiredOrdering(): Array[SortOrder] = {
+ Array[SortOrder](
+ LogicalExpressions.sort(
+ clusterRef,
+ SortDirection.ASCENDING,
+ SortDirection.ASCENDING.defaultNullOrdering())
+ )
+ }
+
+ override def toBatch: DeltaBatchWrite =
+ new TestBatchWrite with DeltaBatchWrite {
+ override def createBatchWriterFactory(
+ info: PhysicalWriteInfo): DeltaWriterFactory = {
+ new DeltaBufferedRowsWriterFactory(capturedWriteSchema)
+ }
+
+ // For column-update writes, updated rows contain only the
declared columns
+ // (columnUpdateSchema from LogicalWriteInfo). We expand each
row to the full
+ // table schema by overlaying them on the base row found by pk.
+ override protected def doCommit(messages:
Array[WriterCommitMessage]): Unit =
+ dataMap.synchronized {
+ val newData = messages.map(_.asInstanceOf[BufferedRows])
+ val writeSchema = capturedWriteSchema
+ val writeFieldIdx =
writeSchema.fieldNames.zipWithIndex.toMap
+
+ val mergedData = newData.map { buf =>
+ val merged = new BufferedRows(buf.key, schema)
+ val updateOpName = UTF8String.fromString(Update.toString)
+ val insertOpName = UTF8String.fromString(Insert.toString)
+ buf.log.foreach { logRow =>
+ val opName = logRow.getUTF8String(0)
+ if (opName == updateOpName) {
+ val pk = logRow.getInt(1)
+ val narrowRow = logRow.get(3,
writeSchema).asInstanceOf[InternalRow]
+ val baseRow = currentRowByPk(pk)
+ val fullRow = overlayNarrowRow(baseRow, narrowRow,
writeFieldIdx)
+ merged.rows.append(fullRow)
+ } else if (opName == insertOpName) {
+ // INSERT rows arrive with the full table schema via
writer.insert()
+ val insertRow = logRow.get(3,
schema).asInstanceOf[InternalRow]
+ merged.rows.append(insertRow.copy())
+ }
+ }
+ merged
+ }
+
+ withDeletes(newData)
+ withData(mergedData)
+ lastWriteLog = newData.flatMap(buffer =>
buffer.log).toIndexedSeq
+ }
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit
= {}
+ }
+ }
+ }
+ }
+ }
+
+ class DeltaBasedColumnUpdateOperationWithReqAttrs(
+ command: Command,
+ reqCols: Array[String],
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty())
+ extends DeltaBasedColumnUpdateOperation(command, options = options) {
+ override def requiredDataAttributes(): Array[NamedReference] =
reqCols.map(FieldReference(_))
+ }
+
+ class DeltaBasedColumnUpdateSplitOperation(
+ command: Command,
+ updatedCols: Seq[NamedReference] = Nil,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty())
+ extends DeltaBasedColumnUpdateOperation(command, updatedCols, options) {
+ override def representUpdateAsDeleteAndInsert(): Boolean = true
+
+ override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder = {
+ lastWriteInfo = info
+ // Capture info into a local val so nested writer/commit closures see a
stable schema
+ // even if a subsequent newWriteBuilder call mutates lastWriteInfo.
+ val capturedInfo = info
+ val capturedWriteSchema = if
(capturedInfo.columnUpdateSchema().isPresent) {
+ capturedInfo.columnUpdateSchema().get()
+ } else {
+ capturedInfo.schema()
+ }
+ new DeltaWriteBuilder {
+ override def build(): DeltaWrite =
+ new DeltaWrite with RequiresDistributionAndOrdering {
+ override def requiredDistribution(): Distribution =
+ Distributions.clustered(Array(PARTITION_COLUMN_REF))
+ override def requiredOrdering(): Array[SortOrder] =
Array[SortOrder](
+ LogicalExpressions.sort(
+ PARTITION_COLUMN_REF,
+ SortDirection.ASCENDING,
+ SortDirection.ASCENDING.defaultNullOrdering()))
+ override def toBatch: DeltaBatchWrite =
+ new TestBatchWrite with DeltaBatchWrite {
+ override def createBatchWriterFactory(
+ info: PhysicalWriteInfo): DeltaWriterFactory = {
+ new DeltaBufferedRowsWriterFactory(capturedWriteSchema)
+ }
+
+ // For delete+reinsert with narrow writes, the REINSERT row
has only the
+ // connector-declared columns (requiredDataAttributes order).
The base row is
+ // looked up by `pk`, so `pk` must be declared for the write
to commit.
+ // Reconstruct the full row by overlaying the narrow row onto
the original.
+ override protected def doCommit(messages:
Array[WriterCommitMessage]): Unit =
+ dataMap.synchronized {
+ val newData = messages.map(_.asInstanceOf[BufferedRows])
+ val writeSchema = capturedWriteSchema
+ val writeFieldIdx =
writeSchema.fieldNames.zipWithIndex.toMap
+ val reinsertOpName =
UTF8String.fromString(Reinsert.toString)
+ val insertOpName = UTF8String.fromString(Insert.toString)
+ val pkIdx = writeFieldIdx("pk")
Review Comment:
**23. Every row-level DELETE on a split column-update table fails at commit,
so the DELETE fallback is not tested on that operation.** Unlike the parent,
which takes `pk` from the row ID inside its `Update` branch (L427-428), this
lookup runs before the log loop. A delta DELETE has no row projection, so
`schema()` is empty and `columnUpdateSchema()` is absent
(`V2Writes.scala:240-241`), `capturedWriteSchema` (L474-478) is empty, and this
line throws `NoSuchElementException: key not found: pk`. Any DELETE that is not
metadata-only, e.g. `DELETE FROM t WHERE pk < 2`, fails this way on
`column-update-split`, `column-update-split-req-attrs` and
`column-update-split-row-id` tables. This is the same line as item 14, but a
different case: item 14 is a split UPDATE that cannot commit.
The round-4 comment
([r4074793068](https://github.com/apache/spark/pull/55518#discussion_r4074793068))
asked for the DELETE and MERGE fallback to be tested on both connector types,
and the PR description lists "DELETE and MERGE falling back to full-width rows"
among what the suites cover. The split operation got "column-update split:
MERGE falls back to full-width rows"
(`DeltaBasedColumnUpdateTableSuite.scala:225`), but no DELETE test. Could the
`pk` lookup move into the `Reinsert` branch, and could a "column-update split:
row-level DELETE falls back to full-width rows" test be added?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedColumnUpdateTableSuite.scala:
##########
@@ -0,0 +1,754 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.{SparkRuntimeException,
SparkUnsupportedOperationException}
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.optimizer.ColumnPruning
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.connector.catalog.{InMemoryBaseTable,
InMemoryRowLevelOperationTable}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+
+class GroupBasedColumnUpdateTableSuite extends UpdateTableSuiteBase {
Review Comment:
**24. Three inherited UPDATE tests do not run against the narrow fixture.**
Since finding 25
([r3741376793](https://github.com/apache/spark/pull/55518#discussion_r3741376793)),
this suite and `DeltaBasedColumnUpdateTableSuite` extend the existing UPDATE
bases, and the PR description says "so the existing UPDATE tests also run
against narrow writes". But `extraTableProps` reaches a table only through the
`createTable` helper (`RowLevelOperationSuiteBase.scala:136`), and three
inherited tests create their table another way, so both suites run them on a
plain table:
- "update table with expression-based default values" (`catalog.createTable`
with a `TableInfo` without properties, `UpdateTableSuiteBase.scala:100-101`)
- "update with current_timestamp default value using DEFAULT keyword"
(`CREATE TABLE`, `UpdateTableSuiteBase.scala:849-850`)
- "update table with recursive CTE" (`CREATE TABLE`,
`UpdateTableSuiteBase.scala:916-917`)
The first one, `SET value = DEFAULT` on a column with an existence default,
is worth running narrow. The fixtures also declare `pk` and `dep`
unconditionally (`InMemoryRowLevelOperationTable.scala:369`, `:574`), so the
CTE table `(val INT)` would fail with
`COLUMN_UPDATE_UNKNOWN_REQUIRED_DATA_ATTRIBUTE` if the properties did reach it.
Could these tables pick up `extraTableProps`, with the fixtures declaring only
existing columns, or could the description say which inherited tests stay on
the wide path?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala:
##########
@@ -228,4 +240,114 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
val expandOutput = generateExpandOutput(attrs, outputs)
Expand(outputs, expandOutput, matchedRowsPlan)
}
+
+ /**
+ * Returns the output of the write relation, which is the layout of the data
rows the connector
+ * receives. If the operation opts into column updates, this is the
connector's
+ * `requiredDataAttributes()`, validated against the assignments. Otherwise
it is the full
+ * relation output.
+ */
+ private def resolveWriteAttrs(
+ relation: DataSourceV2Relation,
+ operation: RowLevelOperation,
+ assignments: Seq[Assignment]): Seq[AttributeReference] = operation match
{
+ case op: SupportsColumnUpdates with SupportsDelta if
op.representUpdateAsDeleteAndInsert =>
+ val attrs = resolveDeclaredAttrs(relation, op, assignments)
+ val rowIdAttrs = resolveRowIdAttrs(relation, op)
+ validateNoRowIdReassignment(op, relation, attrs, assignments, rowIdAttrs)
+ validateRowIdDeclared(op, attrs, rowIdAttrs,
resolveRequiredMetadataAttrs(relation, op))
+ attrs
+ case op: SupportsColumnUpdates =>
+ resolveDeclaredAttrs(relation, op, assignments)
+ case _ =>
+ relation.output
+ }
+
+ /**
+ * Resolves the connector's `requiredDataAttributes()` and checks that it
covers every assigned
+ * column.
+ */
+ private def resolveDeclaredAttrs(
+ relation: DataSourceV2Relation,
+ operation: SupportsColumnUpdates,
+ assignments: Seq[Assignment]): Seq[AttributeReference] = {
+ val attrs = resolveRequiredDataAttrs(relation, operation)
+ validateUpdatedColumnsSubset(operation, assignments, attrs)
+ attrs
+ }
+
+ /**
+ * Enforces that every column being assigned (non-identity) is present in
the connector-declared
+ * `requiredDataAttributes()`. Comparison is at root-column granularity.
+ */
+ private def validateUpdatedColumnsSubset(
+ operation: RowLevelOperation,
+ assignments: Seq[Assignment],
+ connectorDataAttrs: Seq[AttributeReference]): Unit = {
+ val declaredIds = connectorDataAttrs.map(_.exprId).toSet
+ val missing = assignments.collect {
+ case Assignment(key: AttributeReference, value)
+ if !isIdentityAssignment(key, value) &&
!declaredIds.contains(key.exprId) =>
+ key.name
+ }.distinct
+ if (missing.nonEmpty) {
+ throw
QueryCompilationErrors.requiredDataAttributesMissingUpdatedColumnsError(
+ operation.getClass.getName, missing)
+ }
+ }
+
+ /**
+ * For connectors that opt into narrow column updates AND represent UPDATE
as delete + insert,
+ * reject reassignment of any row-ID column, unless
`requiredDataAttributes()` covers every
+ * column in the relation. In that case the REINSERT payload already is the
full row with the
+ * new row-ID value, and the DELETE half still carries the original row-ID
via
+ * `newLazyRowIdProjection`, so reassignment is safe. Otherwise the REINSERT
path has no
+ * row-ID channel to reconstruct columns outside `requiredDataAttributes()`.
+ */
+ private def validateNoRowIdReassignment(
+ operation: RowLevelOperation,
+ relation: DataSourceV2Relation,
+ connectorDataAttrs: Seq[AttributeReference],
+ assignments: Seq[Assignment],
+ rowIdAttrs: Seq[Attribute]): Unit = {
+ val declaredIds = connectorDataAttrs.map(_.exprId).toSet
+ if (relation.output.forall(a => declaredIds.contains(a.exprId))) {
Review Comment:
**21. This exemption never applies when the target relation carries metadata
columns.** `relation.output` is not always the data columns. When an UPDATE is
not resolved in the first iteration of the Resolution batch, for example
because a comparison needs type coercion, `AddMetadataColumns`
(`Analyzer.scala:691`, `:1145`) runs on the resolved command in the next
iteration and calls `withMetadataColumns()` on the target relation
(`Analyzer.scala:1194`), which appends every metadata column to its output
(`DataSourceV2Relation.scala:137-141`). Metadata columns cannot be declared
(`RewriteRowLevelCommand.scala:157`), so this `forall` is then false even when
every data column is declared.
With the setup of "row-ID reassignment is allowed when
requiredDataAttributes covers every table column"
(`DeltaBasedColumnUpdateTableSuite.scala:659`), `UPDATE t SET pk = pk + 10,
salary = -1 WHERE dep = 'hr' AND _partition = 'hr' AND salary > 0.5` fails with
`COLUMN_UPDATE_SPLIT_ROW_ID_REASSIGNMENT`, while the same statement with
`salary > 0`, or without `_partition = 'hr'`, is exempted. This is the
exemption added for finding 22
([r3741376785](https://github.com/apache/spark/pull/55518#discussion_r3741376785)).
Could this compare only the data columns, e.g. `relation.output.filterNot(a
=> MetadataAttribute.isValid(a.metadata))`?
--
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]
