sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3942085900
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,1789 @@ +/* + * 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.comet.contrib.delta + +import java.io.IOException +import java.net.URI +import java.util.Locale + +import scala.collection.mutable.{ListBuffer, Map => MutableMap} +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName} +import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData} +import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues +import org.apache.spark.sql.comet.CometScanExec +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec, SparkPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType} + +import org.apache.comet.CometConf +import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.parquet.CometParquetUtils +import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker} +import org.apache.comet.serde.operator.CometNativeScan +import org.apache.comet.shims.ShimFileFormat + +/** + * Claim/decline gates for the native Delta scan. Correctness rule: when in doubt, decline, + * Spark's Delta reader handles the scan and results stay correct, just unaccelerated. + */ +object DeltaScanSupport { + + /** + * Reader features the native path understands; anything else on the protocol declines the + * table. `deletionVectors`/`columnMapping` are declined separately below for specific reasons. + */ + private val understoodReaderFeatures: Set[String] = + Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck") + + /** + * Is this exactly Delta's DSv1 parquet format? Compared by class name, not `classOf`: a + * `classOf` reference would raise `NoClassDefFoundError` and break every parquet scan when + * delta-spark is absent from the classpath. + */ + def isDeltaScan(scanExec: FileSourceScanExec): Boolean = + scanExec.relation.fileFormat.getClass.getName == + "org.apache.spark.sql.delta.DeltaParquetFileFormat" + + /** + * Claim-time artifacts [[declineReason]] already computes but [[CometDeltaNativeScan.convert]] + * also needs -- threaded through by reference (populated only on the claimable path, right + * before `declineReason` returns `None`) so a claimed scan does not pay to recompute either: + * the Hadoop conf ([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is + * not cheap) and the deletion-vector descriptors (base64-decoded, non-trivial only for DV-shape + * scans). One instance is created per claim attempt in `DeltaScanContrib` and passed to both + * `declineReason` and `convert`. + */ + private[delta] final class DeltaClaimMemo { + var hadoopConf: Configuration = _ + var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty + } + + /** + * First reason this Delta scan cannot go native, or None when claimable (in which case `memo` + * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called when [[isDeltaScan]] + * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert` on a claim, reused + * for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaClaimMemo): Option[String] = { + val format = scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + val protocol = format.protocol + val metadata = format.metadata + // Name mode is supported via physical-name schemas; id mode needs the field-id path and + // stays declined until validated. Hoisted here since several gates below reuse it. + val cmMode = metadata.columnMappingMode.name + // Descriptor deserialization is expensive, so hoist it into a `lazy val`, forced at most + // once in this method; on the claimable path the result is handed to `convert` through + // `memo` below, so a claimed scan deserializes the descriptors exactly once end to end. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates (unsigned-small-int + // fallback, collation, shredded-variant-struct) apply identically here. Pure in-memory check, + // so it runs first, ahead of every I/O-bearing gate below. + val schemaFallbackReasons = new ListBuffer[String]() + val typeChecker = CometScanTypeChecker() + val requiredSchemaSupported = + typeChecker.isSchemaSupported(scanExec.requiredSchema, schemaFallbackReasons) + val partitionSchemaSupported = + typeChecker.isSchemaSupported(scanExec.relation.partitionSchema, schemaFallbackReasons) + if (!requiredSchemaSupported || !partitionSchemaSupported) { + return Some( + "Native Delta scan does not support the schema: " + schemaFallbackReasons.mkString(", ")) + } + + if (format.isCDCRead) { + return Some("Native Delta scan does not support Change Data Feed reads") + } + + // Delta's DML machinery (findTouchedFiles) disables reader optimizations and needs real + // row indexes from Spark's reader; claiming here would feed NULL indexes into DV construction. + if (!format.optimizationsEnabled) { + return Some("Native Delta scan does not support reads with reader optimizations disabled") + } + if (scanExec.requiredSchema.exists(_.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) || + scanExec.relation.dataSchema.exists( + _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) { + return Some("Native Delta scan does not support Delta's generated row-index column") + } + + if (cmMode != "none" && cmMode != "name") { + return Some(s"Native Delta scan does not support column mapping mode $cmMode") + } + // createPhysicalSchema wholesale-replaces field metadata, silently dropping EXISTS_DEFAULT. + if (cmMode == "name" && + getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with column mapping") + } + // createPhysicalSchema rewrites nested StructField names too, and the native builder emits the + // required schema verbatim as output, so name-sensitive expressions (e.g. to_json) would leak + // physical names. Decline until a rename adapter exists. + if (cmMode == "name" && + scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) { + return Some("Native Delta scan does not support column mapping with nested struct fields") + } + + val readerFeatures = protocol.readerFeatureNames + val unknownFeatures = readerFeatures -- understoodReaderFeatures + if (unknownFeatures.nonEmpty) { + return Some( + s"Native Delta scan does not support reader feature(s) ${unknownFeatures.mkString(", ")}") + } + + // Non-constant metadata columns are generated per-row by Spark's reader and unsupported, + // except Delta's DV bookkeeping columns, which the native path emits as constants. + val knownColNames = + scanExec.relation.dataSchema.map(_.name).toSet ++ + scanExec.relation.partitionSchema.map(_.name).toSet ++ + scanExec.fileConstantMetadataColumns.map(_.name).toSet ++ + CometDeltaNativeScan.internalColumnNames + val unknownOutput = scanExec.output.map(_.name).filterNot(knownColNames.contains) + if (unknownOutput.nonEmpty) { + return Some( + s"Native Delta scan does not support generated column(s) ${unknownOutput.mkString(", ")}") + } + + // Deletion-vector shape invariants (see CometDeltaNativeScan.buildDvScanCommon). + if (CometDeltaNativeScan.isDvShape(scanExec)) { + // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping (real row indexes), + // not a DV read; claiming it with a constant would corrupt the DVs being written. + val hasIsRowDeleted = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.IsRowDeletedColumn) + val hasRowIndex = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.RowIndexColumn) + if (hasRowIndex && !hasIsRowDeleted) { + return Some( + "Native Delta scan does not support row-index reads outside a deletion-vector scan") + } + // Internal columns must form a suffix of the read schema so data-column positions agree + // between Spark's output and the stripped native schema. + val names = scanExec.requiredSchema.fields.map(_.name) + val firstInternal = names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains) + if (!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains)) { + return Some("Native Delta scan requires DV bookkeeping columns to trail the read schema") + } + // Native applies the DV itself and emits a dead constant for row-index, so the real value + // must be provably unused above the scan. + if (!rowIndexUnusedAbove(plan, scanExec)) { + return Some( + "Native Delta scan cannot supply _metadata.row_index values consumed by the query") + } + // The DV common builder does not serialize existence defaults yet. + if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with deletion vectors") + } + // Bounds native's memory for expanded DV row selectors (delta_dv.rs), pessimistically + // bounded by 2*cardinality + #row-groups; the conf below makes an over-pessimistic decline + // recoverable. + val maxDeletedRowsPerFile = DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get() + val oversizedCardinalities = dvDescriptors + .map(_.cardinality) + .filter(_ > maxDeletedRowsPerFile) + if (oversizedCardinalities.nonEmpty) { + return Some( + "Native Delta scan does not support a deletion vector deleting " + + s"${oversizedCardinalities.max} rows in a single file, exceeding " + + s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile") + } + } + + // input_file_name & friends read from a thread-local Spark's FileScanRDD sets; the native scan + // does not populate it, and Delta's DML find-touched-files scans use it (mirrors core's check + // in CometScanRule.nativeScan). + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return Some( + "Native Delta scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + + // Row-index metadata columns are generated per-row by Spark's reader (mirrors core); the DV + // shape's trailing row-index column is exempt since the gates above already proved it dead. + if (!CometDeltaNativeScan.isDvShape(scanExec) && + ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema) >= 0) { + return Some("Native Delta scan does not support row index generation") + } + + // Mirror core's vectorized-reader compatibility gate. + if (!SQLConf.get.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED) && + !CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.get()) { + return Some( + "Native Delta scan is incompatible with " + + s"${SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key}=false") + } + + // Decline ALL encrypted-parquet configurations (stricter than core): the exec node does not + // yet wire the decryption-key broadcast to executors. + val hadoopConf = scanExec.relation.sparkSession.sessionState + .newHadoopConfWithOptions(scanExec.relation.options) + // Populated now (rather than only at the very end) so it is available even though several + // early-return gates below still lie ahead: cheap to set, and every one of those gates + // declines the scan anyway, so `memo` is simply never read by `convert` in that case. + memo.hadoopConf = hadoopConf + if (CometParquetUtils.encryptionEnabled(hadoopConf)) { + return Some("Native Delta scan does not support encrypted parquet") + } + + // Nested-type column defaults cannot be serialized; a dropped default would misalign the + // value/index lists consumed positionally on the native side. Mirrors core's + // transformV1Scan gate. + val possibleDefaultValues = getExistenceDefaultValues(scanExec.requiredSchema) + if (possibleDefaultValues.exists(d => + d != null && (d.isInstanceOf[ArrayBasedMapData] || d + .isInstanceOf[GenericInternalRow] || d.isInstanceOf[GenericArrayData]))) { + return Some("Native Delta scan does not support default values for nested types") + } + + // An opted-in S3-compliant alias scheme (fs.comet.s3Compliant.schemes) is declined before the + // generic scheme gate below so the reason says why: core's native scan reads it through the + // S3 client, but the S3 divergence gates further down model Hadoop's S3AFileSystem only. + val aliasReason = + s3CompliantAliasSchemeReason(hadoopConf, scanExec.relation.location.rootPaths.map(_.toUri)) + if (aliasReason.isDefined) { + return aliasReason + } + + // Only claim scans whose root paths object_store (or the configured libhdfs schemes) can + // actually read (mirrors core's unsupportedFsSchemes gate). + val libhdfs = libhdfsSchemes + val unsupportedRootSchemes = + unsupportedSchemes(scanExec.relation.location.rootPaths.map(_.toUri), libhdfs) + if (unsupportedRootSchemes.nonEmpty) { + return Some( + "Native Delta scan does not support filesystem scheme(s) " + + s"${unsupportedRootSchemes.mkString(", ")}") + } + + // A shallow clone can span multiple object-store authorities, but the native builder resolves + // ObjectStoreUrl from only the FIRST selected file; force file listing and decline rather than + // risk reading a later file through the wrong handle. + val dataFileUris = + scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq + + // Both gates below need the DV absolute-path URIs; dvDescriptors is already memoized. + val dvUris = dvDescriptors + .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER) + .map(_.absolutePath(new Path(tableRoot)).toUri) + + // The root-path gate above only inspects the table root(s); selected files can resolve + // through a different scheme (e.g. `viewfs:`). Checked before the authority gates below, + // which presume every URI is natively resolvable. + val unsupportedSelected = unsupportedSelectedSchemeReason(dataFileUris ++ dvUris, libhdfs) + if (unsupportedSelected.isDefined) { + return unsupportedSelected + } + + // Checked before multiStoreReason, which presumes every URI resolves to a single store + // identity -- a userinfo-bearing authority provably does not (store keying drops userinfo). + val userInfoReason = userInfoBearingAuthorityReason(dataFileUris ++ dvUris) + if (userInfoReason.isDefined) { + return userInfoReason + } + + val multiStore = multiStoreReason(dataFileUris) + if (multiStore.isDefined) { + return multiStore + } + + // GCS's zero-I/O, conf-only credential-forwarding gate; ordered alongside the S3 credential + // gates below since all presume a single, well-formed store identity per URI. + val gcsAuthReason = gcsHadoopOnlyAuthReason(hadoopConf, dataFileUris ++ dvUris) + if (gcsAuthReason.isDefined) { + return gcsAuthReason + } + + // Zero-I/O, conf-only, like the GCS gate above: decline any bucket configured for an + // encryption algorithm outside the allowlist (SSE-C, CSE-KMS, CSE-CUSTOM, or unknown) before + // the credential-divergence gates below, which do not otherwise notice this table is readable + // through Hadoop only because Hadoop's request factory (SSE-C) or SDK-level decryption layer + // (CSE-*) does something native never learns about. + val encryptionReason = + unsupportedEncryptionAlgorithmReason(hadoopConf, dataFileUris ++ dvUris) + if (encryptionReason.isDefined) { + return encryptionReason + } + + // Shared across the two gates below: propagateBucketOptions is a full Configuration deep + // copy, and both gates would otherwise recompute it independently for the same bucket(s) + // (once here, then again per-key inside s3ConfigDivergenceReason). One cache, populated + // lazily per bucket on first use, makes it a single copy total per bucket across both gates. + val propagatedConfCache = MutableMap.empty[String, Configuration] + + // Always zero-I/O (plain propagated-conf read, no keystore): native's S3 client has no + // HTTP proxy support at all (no fs.s3a.proxy.* key is read anywhere in s3.rs), so a bucket + // requiring a proxy for S3 egress must decline here rather than claim and then connect + // directly, bypassing whatever network-segmentation/firewall policy required the proxy. + val proxyReason = proxyGateReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache) + if (proxyReason.isDefined) { + return proxyReason + } + + // Zero-I/O, conf-only, like the proxy gate above: Hadoop's AssumedRoleCredentialProvider + // sends fs.s3a.assumed.role.policy as the session policy of its STS AssumeRole request, + // while native's assumed-role provider never reads the key -- a claimed scan would assume + // the role WITHOUT the configured session restriction, silently widening permissions. + val rolePolicyReason = + assumedRolePolicyGateReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache) + if (rolePolicyReason.isDefined) { + return rolePolicyReason + } + + // Every fs.s3a.* option native's get_config (s3.rs) resolves must agree between what Hadoop + // itself would use and what native would read from the forwarded, substituted conf (covers + // long-form bucket credentials, JCEKS/credential-provider shadowing, and any other + // short-vs-effective divergence in one mechanism); reuses hadoopConf from the encryption gate + // above. + val s3Reason = + s3ConfigDivergenceReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache) + if (s3Reason.isDefined) { + return s3Reason + } + + // A credential-provider class native's build_aws_credential_provider_metadata (s3.rs) does + // not recognize errors at scan EXECUTION time, after the scan was already claimed; decline + // eagerly instead. + val providerReason = providerClassGateReason(hadoopConf, dataFileUris ++ dvUris) + if (providerReason.isDefined) { + return providerReason + } + + // Reuse core's generic native-scan gates (ignoreCorruptFiles/ignoreMissingFiles, AQE DPP on + // Spark 3.4, exec enabled); tags its own fallback reasons. + if (!CometNativeScan.isSupported(scanExec)) { + return Some("Core native scan gates rejected the scan (see reasons above)") + } + + // Claimable: hand the already-forced descriptors to `convert` via `memo` so it does not + // deserialize them a second time. + memo.dvDescriptors = dvDescriptors + None + } + + /** + * Deletion-vector descriptors for every file this DV-shape scan selected, normalized to + * absolute on-disk paths. Returns `Seq.empty` for the plain shape. Shared by the DV cardinality + * gate and [[CometDeltaNativeScan.convert]]'s object-store option merge. + */ + private[delta] def selectedDvDescriptors( + scanHelper: CometScanExec, + tableRoot: String): Seq[DeletionVectorDescriptor] = { + if (!CometDeltaNativeScan.isDvShape(scanHelper.wrapped)) { + return Seq.empty + } + val tableRootPath = new Path(tableRoot) + scanHelper.selectedPartitions.iterator + .flatMap(_.files) + .flatMap { file => + file.metadata + .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED) + .map(enc => DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String])) + } + .map(_.copyWithAbsolutePath(tableRootPath)) + .toSeq + } + + /** + * The libhdfs scheme exemption set from [[org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES]], + * parsed exactly like core's scan gate (`NativeConfig.parseSchemeSet`: split on commas, + * trimmed, lowercased) and defaulting to `Set("hdfs")` when unset. + */ + private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match { + case Some(s) => NativeConfig.parseSchemeSet(s) + case None => Set("hdfs") + } + + /** + * Decline reason when any of `uris` uses a scheme opted in as an S3-compliant alias through + * `fs.comet.s3Compliant.schemes` (e.g. `blob`), or `None`. Core's native Parquet scan admits + * such a scheme and reads it through its S3 client, with `NativeConfig` translating the vendor + * `fs.<scheme>.<authority>.*` keys into `fs.s3a.bucket.*` options. Spark, however, reads the + * same table through the vendor's own Hadoop FileSystem, not `S3AFileSystem`, and every S3 + * divergence gate in this object ([[s3ConfigDivergenceReason]] and its siblings) is verified + * against `S3AFileSystem`'s consumers only. With no model of how the vendor filesystem resolves + * its configuration, whether native and Spark would agree cannot be decided, so the scan is + * declined rather than claimed on a guess. Selected data-file and deletion-vector URIs under an + * alias scheme are declined by the generic scheme gates, which never admit an alias (see + * [[unsupportedSchemes]]). + */ + private[delta] def s3CompliantAliasSchemeReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val aliases = NativeConfig.resolveS3CompliantSchemes(hadoopConf) + if (aliases.isEmpty) { + return None + } + val found = uris + .flatMap(uri => Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT))) + .filter(aliases.contains) + .distinct + if (found.isEmpty) { + None + } else { + Some( + "Native Delta scan does not support S3-compliant alias filesystem scheme(s) " + + s"${found.sorted.mkString(", ")} (${CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY}): " + + "Spark reads them through a vendor filesystem whose S3 configuration resolution the " + + "native scan's S3AFileSystem divergence model cannot verify") + } + } + + /** + * The lowercased, deduplicated schemes among `uris` that neither `libhdfs` nor Comet's native + * object_store layer ([[CometScanRule.isNativelyReadableScheme]]) can read. A `null` scheme is + * tolerated, not flagged, since such a URI cannot come from a Hadoop-backed source. The alias + * set handed to core's gate is deliberately empty: an `fs.comet.s3Compliant.schemes` alias is + * never admitted here (see [[s3CompliantAliasSchemeReason]]), even though core admits it. + */ + private[delta] def unsupportedSchemes(uris: Seq[URI], libhdfs: Set[String]): Set[String] = { + uris + .filter { uri => + val sch = uri.getScheme + sch != null && { + val sl = sch.toLowerCase(Locale.ROOT) + !libhdfs.contains(sl) && !CometScanRule.isNativelyReadableScheme(uri, Set.empty) Review Comment: ### Correctness [P2] Preserve the real-path fallback for Delta scans `isNativelyReadableScheme` now probes a synthetic URL such as `file:///`, so this call no longer checks whether object_store accepts the actual path. Core added a separate `objectStoreAcceptsPath` root gate, but Delta returns from the contrib hook before that gate, and `CometNativeScan.isSupported` does not apply it. With native Delta enabled, an otherwise supported local table under a newline-containing directory (URI such as `file:///tmp/dir%0A/data`) is claimed here; the encoded filename reaches native planning, where object_store 0.13.2 decodes `%0A` and rejects the control character. Spark's filesystem reader would have remained usable. The newly introduced case is a fresh JVM with a cold file-scheme cache: the previous helper tested that actual URI and declined it. A previously warmed cache could already hide the old path rejection. Please mirror core's actual-root-path fallback, respecting libhdfs exemptions, and add a Delta regression asserting Spark fallback and the correct answer. This finding is verified from the exact source and locked parser; no full query reproduction was executed. -- 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]
