dongjoon-hyun commented on code in PR #55518:
URL: https://github.com/apache/spark/pull/55518#discussion_r4406298765
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -343,3 +353,63 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
}
+
+object RewriteRowLevelCommand extends SQLConfHelper {
+
+ /**
+ * Resolves the connector-declared required data attributes for
column-update writes against
+ * the given relation, in declared order and with the relation's spelling.
An empty array,
+ * nested fields, duplicate columns, columns that do not exist in the
relation (including a
+ * reference with no name parts) and metadata columns are rejected with an
`AnalysisException`.
+ */
+ private[sql] def resolveRequiredDataAttrs(
+ relation: LogicalPlan,
+ operation: SupportsColumnUpdates): Seq[AttributeReference] = {
+ val refs = operation.requiredDataAttributes.toImmutableArraySeq
+ if (refs.isEmpty) {
Review Comment:
**55. An operation whose declaration is exactly `updatedColumns()` fails an
UPDATE whose assignments are all identity assignments, and the new fixture arms
do not guard against it.** This is my answer to your question in the [latest
summary](https://github.com/apache/spark/pull/55518#pullrequestreview-5672546710).
`collectUpdatedAttrs` skips identity assignments
(`RewriteUpdateTable.scala:252-256`), so `UPDATE t SET salary = salary WHERE pk
= 1` reports no updated columns. An operation that declares exactly
`info.updatedColumns()` then fails here at analysis with
`COLUMN_UPDATE_EMPTY_REQUIRED_DATA_ATTRIBUTES`, while the same statement works
without the mix-in. That declaration is what `SupportsColumnUpdates.java:59-61`
asks for when no lookup or routing column is needed, e.g. for a delta operation
whose row ID arrives through the row ID projection, and the suite's own
`DeltaBasedColumnUpdateSplitOperationWithRowId`
(`InMemoryRowLevelOperationTable.scala:623-631`) is one. It keeps
the mix-in for UPDATE (`InMemoryRowLevelOperationTable.scala:204-207`).
8b1771ad6a7 answers this with a builder-side "should"
(`SupportsColumnUpdates.java:65-66`) and with `updatesNothing` arms in the
default fixtures (`InMemoryRowLevelOperationTable.scala:167-170` and `:190`).
However, `declaredColumns` always starts with `pk`
(`InMemoryRowLevelOperationTable.scala:395-400`), and every table those
fixtures build has `pk`, so the arms prevent no failure. They only switch to a
full-width operation, which the `extraTableProps -> None` arm of the new shared
test (`ColumnUpdateTableSharedTests.scala:235`) then checks. The case `("id =
id, dep = dep", Seq.empty, Seq("pk", "dep"), Row(1, 10, "hr"))`, which passed
on d8a4641, was removed for the same reason.
I'd prefer that Spark not narrow when the declaration is empty, rather than
asking every builder to branch:
- `resolveWriteAttrs` (`RewriteUpdateTable.scala:268-271`) keeps
`validateUpdatedColumnsSubset`, so a non-identity UPDATE with an empty
declaration still fails with
`COLUMN_UPDATE_REQUIRED_DATA_ATTRIBUTES_MISSING_UPDATED_COLUMNS`, and it
returns `relation.output` when the declaration is empty.
- `V2Writes.hasNarrowRows` (`V2Writes.scala:144-146`) uses the same
predicate. Otherwise `validateWriteRelation` (`V2Writes.scala:157-158`)
re-resolves the empty declaration and fails.
- This check, its error condition, the Javadoc at
`SupportsColumnUpdates.java:65-66` and the fixture arms can then go, and the
removed case can come back.
Removing L369-371 alone is not enough. With an empty write relation,
`buildWriteDeltaProjections` returns `rowProjection = None` (L246-251), the
DELETE shape, and `DeltaWritingSparkTask` calls `project` on it for every
UPDATE row (`WriteToDataSourceV2Exec.scala:880` and `:900`). Narrowing only
when `updatedColumns()` is non-empty would also work, but it stops narrowing
identity-only UPDATEs for operations that declare columns (the `Some(Seq("pk",
"salary"))` arm at `ColumnUpdateTableSharedTests.scala:236`), and
`hasNarrowRows` would need `RowLevelOperationTable.operationInfo`, which
`RowLevelWrite` does not expose. If the Javadoc rule stays instead, it needs
the UPDATE scope from [this
comment](https://github.com/apache/spark/pull/55518#discussion_r4395825050),
and it should say that an operation whose declaration can be empty must drop
the mix-in, since keeping the mix-in does not work for such an operation.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -110,8 +111,10 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
// this rule assigns runtime filters to both scan relations (will be
shared at runtime)
// and must transform the runtime filter condition to use correct expr
IDs for each relation
// note this only applies to group-based row-level operations (i.e.
ReplaceData)
+ // the map is built from the original table as the write table may be
narrower than the
Review Comment:
**57.** Nit: this comment reads as if the original table covers every column
the condition references, which does not hold for metadata columns. When the
UPDATE resolves in one analyzer iteration, `AddMetadataColumns` never widens
the target relation, so `write.originalTable.output` has only the data columns,
while a condition such as `_partition = 'hr' AND pk IN (SELECT pk FROM s)`
references `_partition` from `metadataOutput`, which this map does not cover
either. Could the comment say that the map covers the table's data columns,
which the write table may leave out? The metadata-column case behaves the same
on master with `write.table.output`, so I'll look into that part separately.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala:
##########
@@ -228,4 +244,96 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
val expandOutput = generateExpandOutput(attrs, outputs)
Expand(outputs, expandOutput, matchedRowsPlan)
}
+
+ /**
+ * Returns the table attributes that these assignments set to a new value,
skipping identity
+ * assignments.
+ */
+ private def collectUpdatedAttrs(assignments: Seq[Assignment]):
Seq[AttributeReference] = {
+ assignments.collect {
+ case Assignment(key: AttributeReference, value) if
!isIdentityAssignment(key, value) => key
Review Comment:
**56. `updatedColumns()` can name a metadata column, which no operation that
keeps the mix-in can then declare.** This `collect` keeps metadata keys. When
an UPDATE resolves only in a later Resolution iteration, for example because
`salary > 0.5` needs a cast, `AddMetadataColumns` adds the metadata columns to
the target relation (the shape that
`DeltaBasedColumnUpdateTableSuite.scala:450-469` analyzes), and the alignment
over that output (`ResolveRowLevelCommandAssignments.scala:48`) keeps an
assignment to a metadata column. So `UPDATE t SET _partition = 'x' WHERE salary
> 0.5` on a `column-update` or `column-update-cow` table reports
`updatedColumns()` as `[_partition]`:
- The default fixtures follow `SupportsColumnUpdates.java:59-61` and declare
`[pk, dep, _partition]`, which `resolveRequiredDataAttrs` rejects with
`COLUMN_UPDATE_METADATA_REQUIRED_DATA_ATTRIBUTE`
(`RewriteRowLevelCommand.scala:402-408`), as `SupportsColumnUpdates.java:77-80`
says it does.
- An operation that leaves `_partition` out fails
`validateUpdatedColumnsSubset` (L284-287) with
`COLUMN_UPDATE_REQUIRED_DATA_ATTRIBUTES_MISSING_UPDATED_COLUMNS` for
`[_partition]`.
Both messages blame the connector, while on a full-width table the same
statement analyzes and the value goes through the metadata projection. Whether
such an assignment survives the alignment depends on the analyzer iteration
regardless of this PR (with `salary > 0` it is dropped silently), but the dead
end between the two checks is new. Could `collectUpdatedAttrs` skip
`MetadataAttribute` keys, or could the rewrite reject an assignment to a
metadata column with an error that names the user's assignment?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -1500,6 +1501,21 @@ private class BufferWriter(schema: StructType) extends
DataWriter[InternalRow] {
buffer.log.append(logEntry)
}
+ // Tag UPDATE/COPY rows distinctly so tests can verify that Spark dispatches
+ // through writeColumnUpdate (the SupportsColumnUpdates path) rather than
write.
+ override def writeColumnUpdate(metadata: InternalRow, row: InternalRow):
Unit = {
Review Comment:
**58. These overrides are on the `BufferWriter` that every in-memory writer
uses, so a dispatch that sends full-width rows to `writeColumnUpdate` passes
every suite.** A writer that does not override `writeColumnUpdate`, as for any
operation that does not opt in, throws
`DATA_SOURCE_WRITE_COLUMN_UPDATE_NOT_IMPLEMENTED` there
(`DataWriter.java:138-142`). With these overrides, every in-memory writer
appends the row exactly like `write`, and only the log tag differs. The commits
do not look at the tag: `PartitionBasedReplaceData.doCommit` commits
`buffer.rows` (`InMemoryRowLevelOperationTable.scala:269-278`), and
`PartitionBasedNarrowReplaceData.doCommit` overlays each tagged row on its
original row with `columnUpdateSchema().orElse(schema())`
(`InMemoryRowLevelOperationTable.scala:703-704` and `:756-769`), which rebuilds
the same full row. So neither of these regressions would be noticed:
- `DataWithProjectionWritingSparkTask` calls `writeColumnUpdate` regardless
of `useWriteColumnUpdate` (`WriteToDataSourceV2Exec.scala:836-837`).
GroupBasedNoMetadataUpdateTableSuite, GroupBasedNoMetadataDeleteFromTableSuite
and GroupBasedNoMetadataMergeIntoTableSuite do not check the write log, so they
stay green.
- `DataSourceV2Strategy.scala:629` checks only
`isInstanceOf[SupportsColumnUpdates]`, while `V2Writes.hasNarrowRows` keeps
`command == UPDATE` (`V2Writes.scala:144-146`). The DELETE and MERGE rows of a
`column-update-cow` table then go to `writeColumnUpdate`, and
`checkDeleteFallsBackToFullWidthRows` and `checkMergeFallsBackToFullWidthRows`
(`ColumnUpdateTableSharedTests.scala:278-320`) still pass, since they check
only `columnUpdateSchema()`, `updatedColumns()` and the final rows.
Only the metadata-path tests that assert `Write` tags
(`GroupBasedUpdateTableSuite.scala:48-50`,
`GroupBasedDeleteFromTableSuite.scala:47-49` and
`GroupBasedMergeIntoTableSuite.scala:75-81`) would catch the same mistake in
`DataAndMetadataWritingSparkTask`. Could these overrides move to a writer that
only the column-update fixtures create, e.g. a `BufferWriter` subclass used by
`PartitionBasedNarrowReplaceData` and as the delegate of
`RecordOnlyColumnUpdateWriter` (`InMemoryRowLevelOperationTable.scala:911`)?
Alternatively, the two fallback helpers could also assert that the write log
has no `WriteColumnUpdate` entry.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -273,6 +345,440 @@ class InMemoryRowLevelOperationTable private (
override def abort(messages: Array[WriterCommitMessage]): Unit = {}
}
+ // Finds the one current row with the given row-ID values, matching a data
column by value and
+ // `_partition` by the partition key.
+ private def currentRowByRowId(values: Seq[(String, Any)]): InternalRow = {
+ val matches = dataMap.values.iterator.flatten.flatMap { buf =>
+ buf.rows.filter { row =>
+ values.forall {
+ case (name, value) if name == PartitionKeyColumn.name =>
buf.keyString() == value.toString
+ case (name, value) =>
+ val i = schema.fieldIndex(name)
+ row.get(i, schema.fields(i).dataType) == value
+ }
+ }
+ }.toSeq
+ matches match {
+ case Seq(row) => row
+ case _ => throw new IllegalStateException(
+ s"Expected one row with row ID $values, found ${matches.size}")
+ }
+ }
+
+ // Rebuilds a full table row for a narrow column-update row: the base row's
values (or the
+ // existence default for a column added after the base row was written),
overlaid with the
+ // narrow row's columns.
+ private def overlayNarrowRow(
+ baseRow: Option[InternalRow],
+ narrowRow: InternalRow,
+ narrowFieldIdx: Map[String, Int]): InternalRow = {
+ val fullRow = new GenericInternalRow(schema.length)
+ baseRow.foreach { base =>
+ for (i <- schema.fields.indices) {
+ val field = schema.fields(i)
+ val value = if (i < base.numFields) {
+ base.get(i, field.dataType)
Review Comment:
**59. The narrow commits read stored rows by the current schema's position
and type, so they rebuild wrong rows after a column move or a type change.**
`dataMap` keeps each row in the schema it was written with
(`alterTableWithData` keeps the old schema unless a column is dropped,
`InMemoryBaseTable.scala:491-495`), and the reader maps fields by name against
that schema and casts widened types (`InMemoryBaseTable.scala:1312-1313`,
`:1331` and `:1377-1378`). This overlay, `currentRowByRowId` (L356-357) and the
copy-on-write `pk` lookup (L762) use the current ordinal and type instead. For
example, with `(pk INT NOT NULL, salary INT, bonus INT, dep STRING)` and the
rows (1, 100, 10, hr), (2, 200, 20, software) and (3, 300, 30, hr):
- After `ALTER TABLE t ALTER COLUMN salary AFTER bonus`, `UPDATE t SET
salary = -1 WHERE pk = 1` would leave `bonus` = 100 for pk 1 on a
`column-update` table, and `bonus` = 300 for the copied pk 3 on a
`column-update-cow` table.
- After `ALTER COLUMN bonus TYPE DOUBLE`, the overlay reads the stored INT
slot as a DOUBLE, so `bonus` 10 would come back as 4.9E-323 instead of 10.0.
- After moving `pk`, the first narrow UPDATE would fail at commit with
"found 0".
The full-width fixtures keep the right values in all three cases. No test
does this today, but since finding 25 the whole `UpdateTableSuiteBase` battery
runs on these fixtures, so the next schema-evolution UPDATE test added there
would fail only on the narrow suites. Could these helpers read stored rows by
name through the row's own `BufferedRows.schema`, as `BufferedRowsReader` does?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala:
##########
@@ -109,22 +113,98 @@ object V2Writes extends Rule[LogicalPlan] with
PredicateHelper {
WriteToDataSourceV2(Some(r), microBatchWrite, newQuery, customMetrics)
case rd @ ReplaceData(r: DataSourceV2Relation, _, query, _, projections,
_, None) =>
- val rowSchema = projections.rowProjection.schema
+ validateWriteRelation(rd)
+ val (rowSchema, columnUpdateSchema) =
+ writeSchemas(projections.rowProjection.schema, hasNarrowRows(rd))
val metadataSchema = projections.metadataProjection.map(_.schema)
val writeOptions = mergeOptions(Map.empty,
r.options.asCaseSensitiveMap.asScala.toMap)
- val writeBuilder = newWriteBuilder(r.table, writeOptions, rowSchema,
metadataSchema)
+ val writeBuilder = newWriteBuilder(r.table, writeOptions, rowSchema,
metadataSchema,
+ columnUpdateSchema)
val write = writeBuilder.build()
+ validateColumnUpdateWrite(rd, write)
val newQuery = DistributionAndOrderingUtils.prepareQuery(write, query,
r.funCatalog)
rd.copy(write = Some(write), query = newQuery)
case wd @ WriteDelta(r: DataSourceV2Relation, _, query, _, projections, _,
None) =>
+ validateWriteRelation(wd)
val writeOptions = mergeOptions(Map.empty,
r.options.asCaseSensitiveMap.asScala.toMap)
- val deltaWriteBuilder = newDeltaWriteBuilder(r.table, writeOptions,
projections)
+ val deltaWriteBuilder =
+ newDeltaWriteBuilder(r.table, writeOptions, projections,
hasNarrowRows(wd))
val deltaWrite = deltaWriteBuilder.build()
+ validateColumnUpdateWrite(wd, deltaWrite)
val newQuery = DistributionAndOrderingUtils.prepareQuery(deltaWrite,
query, r.funCatalog)
wd.copy(write = Some(deltaWrite), query = newQuery)
}
+ /**
+ * Whether a row-level write delivers narrow rows for updated, copied, and
reinserted records.
+ * This must match the writes whose write relation RewriteUpdateTable
narrows, even when the
+ * declared columns cover the whole table.
+ */
+ private[v2] def hasNarrowRows(write: RowLevelWrite): Boolean = {
+ write.operation.isInstanceOf[SupportsColumnUpdates] &&
write.operation.command == UPDATE
+ }
+
+ /**
+ * Checks that the write relation matches the current operation: a write
that delivers narrow
+ * rows must have exactly the declared columns, in declared order, and any
other write the full
+ * table columns. A rule that replaces the operation after the rewrite can
break this, and the
+ * connector would then decode the rows with the wrong layout.
+ */
+ private def validateWriteRelation(rowLevelWrite: RowLevelWrite): Unit = {
+ val writeNames = rowLevelWrite.table.output.map(_.name)
+ val expectedNames = rowLevelWrite.operation match {
+ case op: SupportsColumnUpdates if hasNarrowRows(rowLevelWrite) =>
+
RewriteRowLevelCommand.resolveRequiredDataAttrs(rowLevelWrite.originalTable, op)
+ .map(_.name)
+ case _ =>
+ rowLevelWrite.originalTable.output.map(_.name)
+ }
+ if (writeNames != expectedNames) {
+ throw SparkException.internalError(
+ s"The write relation of operation
${rowLevelWrite.operation.getClass.getName} has " +
+ s"columns ${writeNames.mkString("[", ", ", "]")}, but the operation
expects " +
+ s"${expectedNames.mkString("[", ", ", "]")}.")
+ }
+ }
+
+ /**
+ * Checks that the distribution and ordering of a write that delivers narrow
rows reference only
+ * columns passed to the write. Other columns stay in the query only until
column pruning removes
+ * them, so they are rejected here whether or not pruning ran.
+ */
+ private def validateColumnUpdateWrite(rowLevelWrite: RowLevelWrite, write:
Write): Unit = {
+ write match {
+ case w: RequiresDistributionAndOrdering if hasNarrowRows(rowLevelWrite)
=>
+ val distribution = w.requiredDistribution match {
+ case d: ClusteredDistribution => d.clustering.toImmutableArraySeq
+ case d: OrderedDistribution => d.ordering.toImmutableArraySeq
+ case _ => Seq.empty
Review Comment:
**61.** Nit: this repeats the distribution match of
`DistributionAndOrderingUtils.prepareQuery` with a different fallback.
`prepareQuery` (`DistributionAndOrderingUtils.scala:41-49`), which runs right
after this check (L124-125 and L134-135), matches `OrderedDistribution`,
`ClusteredDistribution` and `UnspecifiedDistribution` without a wildcard, while
this match ends with `case _ => Seq.empty`. If a distribution kind is added and
only `prepareQuery` learns it, this check would skip that kind's columns, and
an undeclared clustering column would again pass or fail depending on whether
ColumnPruning ran, which item 7 removed. Could both use one helper that returns
the distribution's expressions?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -72,10 +72,12 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
protected def buildOperationTable(
table: SupportsRowLevelOperations,
command: Command,
- options: CaseInsensitiveStringMap): RowLevelOperationTable = {
- val info = RowLevelOperationInfoImpl(command, options)
+ options: CaseInsensitiveStringMap,
+ updatedAttrs: Seq[AttributeReference] = Nil): RowLevelOperationTable = {
+ val updatedColumns = updatedAttrs.map(attr =>
FieldReference(Seq(attr.name)))
Review Comment:
**60. Nothing tests that a top-level column with a dot in its name reaches
`updatedColumns()` as one part.** This line uses
`FieldReference(Seq(attr.name))`, i.e. `FieldReference.column(attr.name)`, so
that a column such as `` `a.b` `` is not parsed into the nested reference ``
`a`.`b` `` ([earlier
thread](https://github.com/apache/spark/pull/55518#discussion_r3634079860)).
However, no `checkLastUpdatedColumns` call uses a dotted name, and the only ``
`a.b` `` UPDATE test (`DeltaBasedColumnUpdateTableSuite.scala:364-375`) runs on
the req-attrs fixture, which ignores `updatedColumns()` and builds its expected
message from `a.name` (`RewriteUpdateTable.scala:284`). So a revert to
`FieldReference(attr.name)` would pass every suite, while connectors would
receive `` `a`.`b` `` again and the default `column-update` fixtures, which
declare the reported columns, would reject ``SET `a.b` = -1`` with
`COLUMN_UPDATE_NESTED_REQUIRED_DATA_ATTRIBUTE`. Could a test on the default
fixture update
`` `a.b` `` and check `` checkLastUpdatedColumns("`a.b`") ``? `describe()`
quotes a dotted part, so the check tells the two forms apart.
`FieldReference.column(attr.name)` would also make the intent visible here.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -128,12 +151,61 @@ class InMemoryRowLevelOperationTable private (
copied.replacedPartitions = replacedPartitions
copied.lastWriteInfo = lastWriteInfo
copied.lastWriteLog = lastWriteLog
+ copied.lastUpdatedColumns = lastUpdatedColumns
copied
}
override def newRowLevelOperationBuilder(
info: RowLevelOperationInfo): RowLevelOperationBuilder = {
- if (properties.getOrDefault(SUPPORTS_DELTAS, "false") == "true") {
+ lastUpdatedColumns = info.updatedColumns()
+ val splitUpdates = properties.getOrDefault(SPLIT_UPDATES,
"false").toBoolean
+ // the copy-on-write commit finds the original row by `pk`
+ val updatesPk = info.updatedColumns().exists(_.describe() == "pk")
+ // The default column-update fixtures drop the mix-in for an UPDATE with
no updated columns,
+ // as SupportsColumnUpdates asks. The fixed-declaration fixtures, the
split row-ID fixture and
+ // all fixtures for DELETE and MERGE keep it.
+ val updatesNothing = info.command == Command.UPDATE &&
info.updatedColumns().isEmpty
+ if (properties.getOrDefault(COLUMN_UPDATE, "false") == "true") {
+ if (updatesNothing) {
+ () => DeltaBasedOperation(info.command, info.options)
+ } else if (splitUpdates) {
+ () => new DeltaBasedColumnUpdateSplitOperation(
+ info.command, info.updatedColumns().toSeq, info.options)
+ } else {
+ () => new DeltaBasedColumnUpdateOperation(
+ info.command, info.updatedColumns().toSeq, info.options)
+ }
+ } else if (properties.containsKey(COLUMN_UPDATE_REQ_ATTRS)) {
+ val reqCols =
properties.get(COLUMN_UPDATE_REQ_ATTRS).split(",").map(_.trim)
+ if (splitUpdates) {
+ () => new DeltaBasedColumnUpdateSplitOperationWithReqAttrs(
+ info.command, reqCols, info.options)
+ } else {
+ () => new DeltaBasedColumnUpdateOperationWithReqAttrs(info.command,
reqCols, info.options)
+ }
+ } else if (properties.getOrDefault(COLUMN_UPDATE_EMPTY_REQ_ATTRS, "false")
== "true") {
Review Comment:
**62.** Nit: `column-update-empty-req-attrs` (L100) builds the same
operation as `column-update-req-attrs` with an empty list. This branch exists
only because `"".split(",")` returns `Array("")` at L179. With
`.filter(_.nonEmpty)` there and at L197, "empty requiredDataAttributes throws
AnalysisException" (`DeltaBasedColumnUpdateTableSuite.scala:336-350`) could use
`reqAttrsTableProps("")`. With item 55's suggestion, this key would go away
anyway.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RowLevelWriteColumnPruningSuite.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.catalyst.optimizer
+
+import java.util
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.ProjectingInternalRow
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, EqualNullSafe, Expression, If, Literal, Not}
+import org.apache.spark.sql.catalyst.plans.{LeftOuter, PlanTest}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Copy, Keep,
Update}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.catalyst.util.{ReplaceDataProjections,
WriteDeltaProjections}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils._
+import org.apache.spark.sql.connector.catalog.{Column, Table, TableCapability}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Tests the [[ColumnPruning]] case that prunes the query columns of a
[[RowLevelWrite]] that its
+ * projections don't read and moves the projections to the ordinals of the
pruned query.
+ */
+class RowLevelWriteColumnPruningSuite extends PlanTest {
+
+ object Optimize extends RuleExecutor[LogicalPlan] {
+ val batches = Batch("Column Pruning", FixedPoint(10),
+ ColumnPruning, CollapseProject, RemoveNoopOperators) :: Nil
+ }
+
+ // t(pk, id, dep, salary, bonus, extra) with the metadata column _partition
+ private val read = LocalRelation(
+ $"pk".int.notNull, $"id".int, $"dep".string, $"salary".int, $"bonus".int,
$"extra".int,
+ $"_partition".string.notNull)
+
+ // the write and original table relations; the pruning never inspects them
+ private val tableRelation: DataSourceV2Relation = {
Review Comment:
**64.** Nit: `TestRelation` could replace this placeholder. As L53 says, the
pruning never inspects these relations, but the placeholder needs an anonymous
`Table`, `DataSourceV2Relation.create` and the imports at L20, L32, L33 and
L35. Catalyst tests already have a minimal `NamedRelation`, `TestRelation`
(`V2WriteAnalysisSuite.scala:137`), which `InlineCTESuite.scala:39` also uses,
so `TestRelation(Nil)` could serve as both `table` and `originalTable`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -273,6 +345,440 @@ class InMemoryRowLevelOperationTable private (
override def abort(messages: Array[WriterCommitMessage]): Unit = {}
}
+ // Finds the one current row with the given row-ID values, matching a data
column by value and
+ // `_partition` by the partition key.
+ private def currentRowByRowId(values: Seq[(String, Any)]): InternalRow = {
+ val matches = dataMap.values.iterator.flatten.flatMap { buf =>
+ buf.rows.filter { row =>
+ values.forall {
+ case (name, value) if name == PartitionKeyColumn.name =>
buf.keyString() == value.toString
+ case (name, value) =>
+ val i = schema.fieldIndex(name)
+ row.get(i, schema.fields(i).dataType) == value
+ }
+ }
+ }.toSeq
+ matches match {
+ case Seq(row) => row
+ case _ => throw new IllegalStateException(
+ s"Expected one row with row ID $values, found ${matches.size}")
+ }
+ }
+
+ // Rebuilds a full table row for a narrow column-update row: the base row's
values (or the
+ // existence default for a column added after the base row was written),
overlaid with the
+ // narrow row's columns.
+ private def overlayNarrowRow(
+ baseRow: Option[InternalRow],
+ narrowRow: InternalRow,
+ narrowFieldIdx: Map[String, Int]): InternalRow = {
+ val fullRow = new GenericInternalRow(schema.length)
+ baseRow.foreach { base =>
+ for (i <- schema.fields.indices) {
+ val field = schema.fields(i)
+ val value = if (i < base.numFields) {
+ base.get(i, field.dataType)
+ } else {
+ ResolveDefaultColumns.getExistenceDefaultValue(field)
+ }
+ fullRow.update(i, value)
+ }
+ }
+ schema.fields.zipWithIndex.foreach { case (field, i) =>
+ narrowFieldIdx.get(field.name).foreach { j =>
+ fullRow.update(i, narrowRow.get(j, field.dataType))
+ }
+ }
+ fullRow
+ }
+
+ // The base columns `pk` and `dep`, where the table has them, followed by
the updated columns.
+ private def declaredColumns(updatedCols: Seq[NamedReference]):
Array[NamedReference] = {
+ val base = Seq("pk",
"dep").filter(schema.fieldNames.contains).map(FieldReference(_))
+ val baseNames = base.map(_.describe()).toSet
+ (base ++ updatedCols.filterNot(r =>
baseNames.contains(r.describe()))).toArray
+ }
+
+ // 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`, where the table has them, 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] =
declaredColumns(updatedCols)
+
+ // The case-class equality of DeltaBasedOperation would make this
operation equal to a wide
+ // one, or to another column-update operation, with the same command and
options.
+ override def equals(other: Any): Boolean = other match {
+ case ref: AnyRef => this eq ref
+ case _ => false
+ }
+ override def hashCode(): Int = System.identityHashCode(this)
+ override def canEqual(other: Any): Boolean = false
+
+ private val clusterExprs: Seq[Expression] = {
+ val refs = Option(properties.get(COLUMN_UPDATE_CLUSTER_BY))
+ .map(_.split(",").map(name => FieldReference(name.trim)).toSeq)
+ .getOrElse(Seq(PARTITION_COLUMN_REF))
+ if (properties.getOrDefault(COLUMN_UPDATE_CLUSTER_BY_NESTED_TRANSFORM,
"false") == "true") {
+ refs.map(ref => LogicalExpressions.apply("f",
LogicalExpressions.apply("g", ref)))
+ } else {
+ refs
+ }
+ }
+
+ override def newWriteBuilder(info: LogicalWriteInfo): DeltaWriteBuilder = {
+ lastWriteInfo = info
+ val writeSchema = info.columnUpdateSchema().orElse(info.schema())
+ val batchWrite = new TestBatchWrite with DeltaBatchWrite {
+ override def createBatchWriterFactory(
+ physicalInfo: PhysicalWriteInfo): DeltaWriterFactory = {
+ newWriterFactory(info, writeSchema)
+ }
+
+ override protected def doCommit(messages: Array[WriterCommitMessage]):
Unit =
+ dataMap.synchronized {
+ val newData = messages.map(_.asInstanceOf[BufferedRows])
+ commitColumnUpdate(info, writeSchema, newData)
+ }
+
+ override def abort(messages: Array[WriterCommitMessage]): Unit = {}
+ }
+ new DeltaWriteBuilder {
+ override def build(): DeltaWrite =
+ if (noMetadata && !properties.containsKey(COLUMN_UPDATE_CLUSTER_BY))
{
+ new DeltaWrite {
+ override def toBatch: DeltaBatchWrite = batchWrite
+ }
+ } else {
+ new DeltaWrite with RequiresDistributionAndOrdering {
+
+ override def requiredDistribution(): Distribution = {
+ Distributions.clustered(clusterExprs.toArray)
+ }
+
+ override def requiredOrdering(): Array[SortOrder] = {
+ clusterExprs.map { expr =>
+ LogicalExpressions.sort(
+ expr,
+ SortDirection.ASCENDING,
+ SortDirection.ASCENDING.defaultNullOrdering()): SortOrder
+ }.toArray
+ }
+
+ override def toBatch: DeltaBatchWrite = batchWrite
+ }
+ }
+ }
+ }
+
+ protected def newWriterFactory(
+ info: LogicalWriteInfo,
+ writeSchema: StructType): DeltaWriterFactory = {
+ new DeltaBufferedRowsWriterFactory(writeSchema)
+ }
+
+ // 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.
+ protected def commitColumnUpdate(
+ info: LogicalWriteInfo,
+ writeSchema: StructType,
+ newData: Array[BufferedRows]): Unit = {
+ val writeFieldIdx = writeSchema.fieldNames.zipWithIndex.toMap
Review Comment:
**65.** Nit: `StructType.getFieldIndex` already provides this name-to-index
map. The three commits build `writeSchema.fieldNames.zipWithIndex.toMap` here,
at L555 and at L751, and pass it to `overlayNarrowRow` (L371-374), which has to
trust that the map belongs to `narrowRow`. Passing the narrow `StructType` and
calling `getFieldIndex` (cached; `BufferedRowsReader` uses it at
`InMemoryBaseTable.scala:1331`) would drop the three copies and tie the index
to the row's schema. The `fieldNames.contains(name)` plus `fieldIndex(name)`
pair at L567-571 would also become one `getFieldIndex` call.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/ColumnUpdateTableSharedTests.scala:
##########
@@ -0,0 +1,387 @@
+/*
+ * 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 scala.jdk.CollectionConverters._
+
+import org.apache.spark.{SparkException, SparkRuntimeException}
+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.connector.write.{RowLevelOperation,
RowLevelOperationTable}
+import org.apache.spark.sql.connector.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
V2Writes}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+// Column-update tests that hold for both delta-based and group-based
operations; each suite
+// supplies its column-update table properties through `extraTableProps`.
+trait ColumnUpdateTableSharedTests extends UpdateTableSuiteBase {
+
+ // the properties of a column-update table whose connector declares exactly
`reqAttrs`
+ protected def reqAttrsTableProps(reqAttrs: String): java.util.Map[String,
String]
+
+ // the properties of a table of the same kind whose operation does not mix
in column updates
+ protected def fullWidthTableProps: java.util.Map[String, String]
+
+ // the metadata columns the scan of a column-update table reads, in DDL form
+ protected def scanMetadataColumns: String = ", _partition STRING, index INT"
+
+ // the class name of the operation that `reqAttrsTableProps` builds
+ protected def reqAttrsOperationClass: String = if (deltaUpdate) {
+
classOf[InMemoryRowLevelOperationTable#DeltaBasedColumnUpdateOperationWithReqAttrs].getName
+ } else {
+
classOf[InMemoryRowLevelOperationTable#PartitionBasedColumnUpdateOperationWithReqAttrs].getName
+ }
+
+ // `base` with the given table properties added; a given key replaces the
same key in `base`
+ protected def tablePropsWith(
+ base: java.util.Map[String, String],
+ props: (String, String)*): java.util.Map[String, String] = {
+ (base.asScala ++ props).asJava
+ }
+
+ // t(pk, id, dep, salary, bonus, extra) with CHECK (extra > 0), partitioned
by dep
+ protected def createAndInitTableWithProps(props: java.util.Map[String,
String]): Unit = {
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING, salary INT, bonus
INT, extra INT",
+ """{ "pk": 1, "id": 1, "dep": "hr", "salary": 100, "bonus": 10, "extra":
1 }
+ |{ "pk": 2, "id": 2, "dep": "hr", "salary": 200, "bonus": 20, "extra":
1 }
+ |{ "pk": 3, "id": 3, "dep": "software", "salary": 300, "bonus": 30,
"extra": 1 }
+ |""".stripMargin,
+ props)
+ sql(s"ALTER TABLE $tableNameAsString ADD CONSTRAINT positive_extra CHECK
(extra > 0)")
+ }
+
+ test("column-update: undeclared columns are not written and keep their
values") {
+ createAndInitTable("pk INT NOT NULL, salary INT, bonus INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "bonus": 10, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "bonus": 20, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "bonus": 30, "dep": "hr" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE dep = 'hr'")
+
+ checkLastColumnUpdateSchema("pk", "dep", "salary")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, -1, 10, "hr") ::
+ Row(2, 200, 20, "software") ::
+ Row(3, -1, 30, "hr") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 2, numCopiedRows = 0)
+ }
+
+ test("column-update: write relation has the declared columns in declared
order") {
+ createAndInitTableWithProps(reqAttrsTableProps("salary,pk"))
+
+ val write = executeAndKeepAnalyzedWrite {
+ sql(s"UPDATE $tableNameAsString SET salary = salary + bonus WHERE dep =
'hr'")
+ }
+
+ assert(!write.query.exists(_.isInstanceOf[Union]))
+ checkColumnUpdateWrite(write, "salary", "pk")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, 1, "hr", 110, 10, 1) ::
+ Row(2, 2, "hr", 220, 20, 1) ::
+ Row(3, 3, "software", 300, 30, 1) :: Nil)
+ }
+
+ test("column-update: the plan writes the declared columns and scans the
referenced ones") {
+ createAndInitTableWithProps(reqAttrsTableProps("pk,salary"))
+
+ // `id` is neither declared nor referenced; `dep`, `bonus` and `extra` are
read for the
+ // condition, the assignment and the CHECK constraint
+ val write = executeAndKeepAnalyzedWrite {
+ executeAndCheckScan(
+ s"UPDATE $tableNameAsString SET salary = salary + bonus WHERE dep =
'hr'",
+ s"pk INT, dep STRING, salary INT, bonus INT, extra
INT$scanMetadataColumns")
+ }
+
+ val writeNodeName = if (deltaUpdate) "WriteDelta" else "ReplaceData"
+ val header = write.simpleString(Int.MaxValue)
+ assert(header.matches(s"""$writeNodeName RelationV2\\[pk#\\d+,
salary#\\d+\\] .*"""), header)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, 1, "hr", 110, 10, 1) ::
+ Row(2, 2, "hr", 220, 20, 1) ::
+ Row(3, 3, "software", 300, 30, 1) :: Nil)
+ }
+
+ test("column-update: undeclared, differently-cased columns in the RHS and
WHERE clause are " +
+ "read with the table's own spelling") {
+ createAndInitTable("pk INT NOT NULL, salary INT, bonus INT, extra INT, dep
STRING",
+ """{ "pk": 1, "salary": 100, "bonus": 10, "extra": 5, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "bonus": 20, "extra": 5, "dep": "hr" }
+ |""".stripMargin)
+
+ executeAndCheckScan(
+ s"UPDATE $tableNameAsString SET salary = salary + BONUS WHERE pk = 1 AND
EXTRA > 3",
+ s"pk INT, salary INT, bonus INT, extra INT, dep
STRING$scanMetadataColumns")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, 110, 10, 5, "hr") :: Row(2, 200, 20, 5, "hr") :: Nil)
+ }
+
+ test("column-update: assigned column outside requiredDataAttributes is
rejected") {
+ createAndInitTableWithProps(reqAttrsTableProps("pk,salary"))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"UPDATE $tableNameAsString SET bonus = 0 WHERE dep = 'hr'")
+ },
+ condition =
"COLUMN_UPDATE_REQUIRED_DATA_ATTRIBUTES_MISSING_UPDATED_COLUMNS",
+ parameters = Map("connector" -> reqAttrsOperationClass, "missingColumns"
-> "[`bonus`]"))
+ }
+
+ test("column-update: an operation replaced after the rewrite is rejected") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |""".stripMargin)
+
+ val write = executeAndKeepAnalyzedWrite {
+ sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE pk = 1")
+ }
+ val relation = write.table.asInstanceOf[DataSourceV2Relation]
+ val operationTable = relation.table.asInstanceOf[RowLevelOperationTable]
+ val t = operationTable.table.asInstanceOf[InMemoryRowLevelOperationTable]
+ val replacements: Seq[(RowLevelOperation, String)] = if (deltaUpdate) {
+ Seq(
+ t.DeltaBasedOperation(UPDATE, CaseInsensitiveStringMap.empty()) ->
"[pk, salary, dep]",
+ new t.DeltaBasedColumnUpdateOperation(UPDATE) -> "[pk, dep]")
+ } else {
+ Seq(
+ t.PartitionBasedOperation(UPDATE, CaseInsensitiveStringMap.empty()) ->
"[pk, salary, dep]",
+ new t.PartitionBasedColumnUpdateOperation(UPDATE) -> "[pk, dep]")
+ }
+ replacements.foreach { case (operation, expectedNames) =>
+ assert(operation != operationTable.operation)
+ val replaced = write.withNewTable(
+ relation.copy(table = operationTable.copy(operation = operation)(
+ operationTable.operationInfo)))
+ checkError(
+ exception = intercept[SparkException](V2Writes(replaced)),
+ condition = "INTERNAL_ERROR",
+ parameters = Map("message" ->
+ (s"The write relation of operation ${operation.getClass.getName} has
columns " +
+ s"[pk, dep, salary], but the operation expects $expectedNames.")))
+ }
+ }
+
+ test("column-update: nested struct field update narrows to the root struct
column") {
+ createAndInitTable("pk INT NOT NULL, s STRUCT<c1: INT, c2: INT>, dep
STRING",
+ """{ "pk": 1, "s": { "c1": 1, "c2": 2 }, "dep": "hr" }
+ |{ "pk": 2, "s": { "c1": 3, "c2": 4 }, "dep": "hr" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET s.c1 = -1 WHERE pk = 1")
+
+ checkLastUpdatedColumns("s")
+ checkLastColumnUpdateSchema("pk", "dep", "s")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, Row(-1, 2), "hr") :: Row(2, Row(3, 4), "hr") :: Nil)
+ }
+
+ test("column-update: excluding ColumnPruning keeps the write narrow") {
+ createAndInitTableWithProps(reqAttrsTableProps("pk,salary"))
+
+ withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
ColumnPruning.ruleName) {
+ executeAndCheckScan(
+ s"UPDATE $tableNameAsString SET salary = salary + bonus WHERE dep =
'hr'",
+ s"pk INT, id INT, dep STRING, salary INT, bonus INT, extra
INT$scanMetadataColumns")
+ }
+
+ checkLastColumnUpdateSchema("pk", "salary")
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk"),
+ Row(1, 1, "hr", 110, 10, 1) ::
+ Row(2, 2, "hr", 220, 20, 1) ::
+ Row(3, 3, "software", 300, 30, 1) :: Nil)
+ }
+
+ test("column-update: reassigning a declared row-ID column keeps the row's
other columns") {
+ createAndInitTableWithProps(reqAttrsTableProps("pk,salary"))
+
+ sql(s"UPDATE $tableNameAsString SET pk = 2, salary = 0 WHERE pk = 1")
+
+ // the copy-on-write fixture finds original rows by `pk`, so it falls back
to full-width rows
+ assert(table.lastWriteInfo.columnUpdateSchema().isPresent == deltaUpdate)
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString ORDER BY pk, id"),
+ Row(2, 1, "hr", 0, 10, 1) ::
+ Row(2, 2, "hr", 200, 20, 1) ::
+ Row(3, 3, "software", 300, 30, 1) :: Nil)
+ }
+
+ test("column-update: an UPDATE whose assignments are all identity
assignments is narrowed " +
+ "only if the operation keeps the mix-in") {
+ Seq(
+ extraTableProps -> None,
+ reqAttrsTableProps("pk,salary") -> Some(Seq("pk", "salary"))
+ ).foreach { case (props, columnUpdateNames) =>
+ withTable(tableNameAsString) {
+ createAndInitTableWithProps(props)
+
+ sql(s"UPDATE $tableNameAsString SET salary = salary, dep = dep WHERE
pk = 1")
+
+ checkLastUpdatedColumns()
+ columnUpdateNames match {
+ case Some(names) => checkLastColumnUpdateSchema(names: _*)
+ case None =>
+ assert(!table.lastWriteInfo.columnUpdateSchema().isPresent,
Review Comment:
**63.** Nit: the full-width check is written out three times with three
messages. L247-248, L281-283 and L312-314 repeat
`assert(!table.lastWriteInfo.columnUpdateSchema().isPresent, ...)`, and L224
checks `isPresent == deltaUpdate` without a message. Could
`RowLevelOperationSuiteBase` get a `checkLastWriteIsFullWidth()` next to
`checkLastColumnUpdateSchema` (item 45), with L224 picking one of the two by
`deltaUpdate`?
--
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]
