This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 54ae6fb8f [spark] supports metadata columns (#3307)
54ae6fb8f is described below
commit 54ae6fb8f12e5b13b2039b012b2e11b9e8e039ba
Author: Yann Byron <[email protected]>
AuthorDate: Sat May 11 13:00:16 2024 +0800
[spark] supports metadata columns (#3307)
---
.../java/org/apache/paimon/spark/SparkTable.java | 11 +++++
.../org/apache/paimon/spark/PaimonBaseScan.scala | 21 +++++++--
.../org/apache/paimon/spark/PaimonBatch.scala | 12 +++--
.../paimon/spark/PaimonPartitionReader.scala | 6 ++-
.../spark/PaimonPartitionReaderFactory.scala | 21 +++++++--
.../paimon/spark/PaimonRecordReaderIterator.scala | 44 +++++++++++++++--
.../org/apache/paimon/spark/PaimonStatistics.scala | 20 ++++----
.../paimon/spark/schema/PaimonMetadataColumn.scala | 55 ++++++++++++++++++++++
.../spark/sources/PaimonMicroBatchStream.scala | 2 +-
.../apache/paimon/spark/sql/PaimonQueryTest.scala | 36 ++++++++++++++
10 files changed, 201 insertions(+), 27 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkTable.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkTable.java
index c319679de..be9f4309a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkTable.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkTable.java
@@ -20,10 +20,13 @@ package org.apache.paimon.spark;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.options.Options;
+import org.apache.paimon.spark.schema.PaimonMetadataColumn$;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
+import org.apache.spark.sql.connector.catalog.MetadataColumn;
+import org.apache.spark.sql.connector.catalog.SupportsMetadataColumns;
import org.apache.spark.sql.connector.catalog.SupportsRead;
import org.apache.spark.sql.connector.catalog.SupportsWrite;
import org.apache.spark.sql.connector.catalog.TableCapability;
@@ -49,6 +52,7 @@ public class SparkTable
implements org.apache.spark.sql.connector.catalog.Table,
SupportsRead,
SupportsWrite,
+ SupportsMetadataColumns,
PaimonPartitionManagement {
private final Table table;
@@ -135,4 +139,11 @@ public class SparkTable
SparkTable that = (SparkTable) o;
return Objects.equals(table, that.table);
}
+
+ @Override
+ public MetadataColumn[] metadataColumns() {
+ return new MetadataColumn[] {
+ PaimonMetadataColumn$.MODULE$.FILE_PATH(),
PaimonMetadataColumn$.MODULE$.ROW_INDEX()
+ };
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
index 7a49167a9..f6df901a4 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark
import org.apache.paimon.{stats, CoreOptions}
import org.apache.paimon.predicate.{Predicate, PredicateBuilder}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.spark.sources.PaimonMicroBatchStream
import org.apache.paimon.spark.statistics.StatisticsHelper
import org.apache.paimon.table.{DataTable, FileStoreTable, Table}
@@ -51,6 +52,16 @@ abstract class PaimonBaseScan(
private lazy val tableSchema = SparkTypeUtils.fromPaimonRowType(tableRowType)
+ private val (tableFields, metadataFields) = {
+ val requiredFieldNames = requiredSchema.fieldNames
+ val _tableFields = tableSchema.filter(field =>
requiredFieldNames.contains(field.name))
+ val _metadataFields =
+ requiredSchema
+ .filterNot(field => tableSchema.fieldNames.contains(field.name))
+ .filter(field =>
PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS.contains(field.name))
+ (_tableFields, _metadataFields)
+ }
+
protected var runtimeFilters: Array[Filter] = Array.empty
protected var splits: Array[Split] = _
@@ -60,14 +71,14 @@ abstract class PaimonBaseScan(
lazy val statistics: Optional[stats.Statistics] = table.statistics()
lazy val requiredStatsSchema: StructType = {
- val fieldNames = requiredSchema.fieldNames ++
reservedFilters.flatMap(_.references)
+ val fieldNames = tableFields.map(_.name) ++
reservedFilters.flatMap(_.references)
StructType(tableSchema.filter(field => fieldNames.contains(field.name)))
}
lazy val readBuilder: ReadBuilder = {
val _readBuilder = table.newReadBuilder()
- val projection = readSchema().fieldNames.map(field =>
tableRowType.getFieldNames.indexOf(field))
+ val projection = tableFields.map(field =>
tableSchema.fieldNames.indexOf(field.name)).toArray
_readBuilder.withProjection(projection)
if (filters.nonEmpty) {
val pushedPredicate = PredicateBuilder.and(filters: _*)
@@ -90,12 +101,12 @@ abstract class PaimonBaseScan(
}
override def readSchema(): StructType = {
- val requiredFieldNames = requiredSchema.fieldNames
- StructType(tableSchema.filter(field =>
requiredFieldNames.contains(field.name)))
+ StructType(tableFields ++ metadataFields)
}
override def toBatch: Batch = {
- PaimonBatch(getSplits, readBuilder)
+ val metadataColumns = metadataFields.map(field =>
PaimonMetadataColumn.get(field.name))
+ PaimonBatch(getSplits, readBuilder, metadataColumns)
}
override def toMicroBatchStream(checkpointLocation: String):
MicroBatchStream = {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
index 7544aa2d9..c7b782573 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
@@ -18,20 +18,26 @@
package org.apache.paimon.spark
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.source.{ReadBuilder, Split}
import org.apache.spark.sql.connector.read.{Batch, InputPartition,
PartitionReaderFactory}
+import org.apache.spark.sql.types.StructType
import java.util.Objects
/** A Spark [[Batch]] for paimon. */
-case class PaimonBatch(splits: Array[Split], readBuilder: ReadBuilder) extends
Batch {
+case class PaimonBatch(
+ splits: Array[Split],
+ readBuilder: ReadBuilder,
+ metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty)
+ extends Batch {
override def planInputPartitions(): Array[InputPartition] =
splits.map(new SparkInputPartition(_).asInstanceOf[InputPartition])
- override def createReaderFactory(): PartitionReaderFactory = new
PaimonPartitionReaderFactory(
- readBuilder)
+ override def createReaderFactory(): PartitionReaderFactory =
+ PaimonPartitionReaderFactory(readBuilder, metadataColumns)
override def equals(obj: Any): Boolean = {
obj match {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala
index c4e694814..1509ea83c 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark
import org.apache.paimon.data.{InternalRow => PaimonInternalRow}
import org.apache.paimon.reader.{RecordReader, RecordReaderIterator}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.source.{DataSplit, Split}
import org.apache.spark.sql.catalyst.InternalRow
@@ -33,14 +34,15 @@ import scala.collection.JavaConverters._
case class PaimonPartitionReader(
readFunc: Split => RecordReader[PaimonInternalRow],
partition: SparkInputPartition,
- row: SparkInternalRow
+ row: SparkInternalRow,
+ metadataColumns: Seq[PaimonMetadataColumn]
) extends PartitionReader[InternalRow] {
private lazy val split: Split = partition.split
private lazy val iterator = {
val reader = readFunc(split)
- PaimonRecordReaderIterator(reader)
+ PaimonRecordReaderIterator(reader, metadataColumns)
}
override def next(): Boolean = {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala
index 961aad8a2..3ac724fcc 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala
@@ -22,25 +22,38 @@ import org.apache.paimon.data
import org.apache.paimon.disk.IOManager
import org.apache.paimon.reader.RecordReader
import org.apache.paimon.spark.SparkUtils.createIOManager
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.source.{ReadBuilder, Split}
+import org.apache.paimon.types.RowType
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader,
PartitionReaderFactory}
+import java.util.{ArrayList => JList}
import java.util.Objects
-case class PaimonPartitionReaderFactory(readBuilder: ReadBuilder) extends
PartitionReaderFactory {
+import scala.collection.JavaConverters._
+
+case class PaimonPartitionReaderFactory(
+ readBuilder: ReadBuilder,
+ metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty)
+ extends PartitionReaderFactory {
private lazy val ioManager: IOManager = createIOManager()
- private lazy val row: SparkInternalRow = new
SparkInternalRow(readBuilder.readType())
+ private lazy val row: SparkInternalRow = {
+ val dataFields = new JList(readBuilder.readType().getFields)
+ dataFields.addAll(metadataColumns.map(_.toPaimonDataField).asJava)
+ val rowType = new RowType(dataFields)
+ new SparkInternalRow(rowType)
+ }
override def createReader(partition: InputPartition):
PartitionReader[InternalRow] = {
partition match {
case paimonInputPartition: SparkInputPartition =>
val readFunc: Split => RecordReader[data.InternalRow] =
(split: Split) =>
readBuilder.newRead().withIOManager(ioManager).createReader(split)
- PaimonPartitionReader(readFunc, paimonInputPartition, row)
+ PaimonPartitionReader(readFunc, paimonInputPartition, row,
metadataColumns)
case _ =>
throw new RuntimeException(s"It's not a Paimon input partition,
$partition")
}
@@ -49,7 +62,7 @@ case class PaimonPartitionReaderFactory(readBuilder:
ReadBuilder) extends Partit
override def equals(obj: Any): Boolean = {
obj match {
case other: PaimonPartitionReaderFactory =>
- this.readBuilder.equals(other.readBuilder)
+ this.readBuilder.equals(other.readBuilder) && this.metadataColumns ==
other.metadataColumns
case _ => false
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
index 36fb03fd6..ce880e670 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
@@ -18,23 +18,32 @@
package org.apache.paimon.spark
-import org.apache.paimon.data.{InternalRow => PaimonInternalRow}
+import org.apache.paimon.data.{BinaryString, GenericRow, InternalRow =>
PaimonInternalRow, JoinedRow}
import org.apache.paimon.fs.Path
import org.apache.paimon.reader.{FileRecordIterator, RecordReader}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.utils.CloseableIterator
import org.apache.spark.sql.PaimonUtils
import java.io.IOException
-case class PaimonRecordReaderIterator(reader: RecordReader[PaimonInternalRow])
+case class PaimonRecordReaderIterator(
+ reader: RecordReader[PaimonInternalRow],
+ metadataColumns: Seq[PaimonMetadataColumn])
extends CloseableIterator[PaimonInternalRow] {
private var lastFilePath: Path = _
+ private var isFileRecordIterator: Boolean = false
private var currentIterator: RecordReader.RecordIterator[PaimonInternalRow]
= readBatch()
private var advanced = false
private var currentResult: PaimonInternalRow = _
+ private val needMetadata = metadataColumns.nonEmpty
+ private val metadataRow: GenericRow =
+ GenericRow.of(Array.fill(metadataColumns.size)(null.asInstanceOf[AnyRef]):
_*)
+ private val joinedRow: JoinedRow = JoinedRow.join(null, metadataRow)
+
override def hasNext: Boolean = {
if (currentIterator == null) {
false
@@ -69,11 +78,13 @@ case class PaimonRecordReaderIterator(reader:
RecordReader[PaimonInternalRow])
val iter = reader.readBatch()
iter match {
case fileRecordIterator: FileRecordIterator[_] =>
+ isFileRecordIterator = true
if (lastFilePath != fileRecordIterator.filePath()) {
PaimonUtils.setInputFileName(fileRecordIterator.filePath().toUri.toString)
lastFilePath = fileRecordIterator.filePath()
}
case _ =>
+ isFileRecordIterator = false
}
iter
}
@@ -84,7 +95,22 @@ case class PaimonRecordReaderIterator(reader:
RecordReader[PaimonInternalRow])
try {
var stop = false
while (!stop) {
- currentResult = currentIterator.next
+ val dataRow = currentIterator.next()
+ if (dataRow != null) {
+ if (needMetadata) {
+ if (!isFileRecordIterator) {
+ throw new RuntimeException(
+ "There need be FileRecoredIterator when metadata columns are
required.")
+ }
+
updateMetadataRow(currentIterator.asInstanceOf[FileRecordIterator[PaimonInternalRow]])
+ currentResult = joinedRow.replace(dataRow, metadataRow)
+ } else {
+ currentResult = dataRow
+ }
+ } else {
+ currentResult = null
+ }
+
if (currentResult != null) {
stop = true
} else {
@@ -102,4 +128,16 @@ case class PaimonRecordReaderIterator(reader:
RecordReader[PaimonInternalRow])
}
}
}
+
+ private def updateMetadataRow(fileRecordIterator:
FileRecordIterator[PaimonInternalRow]): Unit = {
+ metadataColumns.zipWithIndex.foreach {
+ case (metadataColumn, index) =>
+ metadataColumn.name match {
+ case PaimonMetadataColumn.ROW_INDEX.name =>
+ metadataRow.setField(index, fileRecordIterator.returnedPosition())
+ case PaimonMetadataColumn.FILE_PATH.name =>
+ metadataRow.setField(index,
BinaryString.fromString(lastFilePath.toUri.toString))
+ }
+ }
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
index abaac6382..fcf386397 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
@@ -48,19 +48,21 @@ case class PaimonStatistics[T <: PaimonBaseScan](scan: T)
extends Statistics {
if (paimonStats.isPresent) paimonStats.get().mergedRecordCount() else
OptionalLong.of(rowCount)
override def columnStats(): java.util.Map[NamedReference, ColumnStatistics]
= {
- val requiredFields = scan.requiredStatsSchema.fieldNames.toList.asJava
+ val requiredFields = scan.requiredStatsSchema.fieldNames
val resultMap = new java.util.HashMap[NamedReference, ColumnStatistics]()
if (paimonStats.isPresent) {
val paimonColStats = paimonStats.get().colStats()
- scan.tableRowType.getFields
- .stream()
- .filter(
- field => requiredFields.contains(field.name) &&
paimonColStats.containsKey(field.name()))
- .forEach(
- f =>
+ scan.tableRowType.getFields.asScala
+ .filter {
+ field => requiredFields.contains(field.name) &&
paimonColStats.containsKey(field.name())
+ }
+ .foreach {
+ field =>
resultMap.put(
- PaimonUtils.fieldReference(f.name()),
- PaimonColumnStats(f.`type`(), paimonColStats.get(f.name()))))
+ PaimonUtils.fieldReference(field.name()),
+ PaimonColumnStats(field.`type`(),
paimonColStats.get(field.name()))
+ )
+ }
}
resultMap
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
new file mode 100644
index 000000000..74ce7090d
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.paimon.spark.schema
+
+import org.apache.paimon.spark.SparkTypeUtils
+import org.apache.paimon.types.DataField
+
+import org.apache.spark.sql.connector.catalog.MetadataColumn
+import org.apache.spark.sql.types.{DataType, LongType, StringType}
+
+case class PaimonMetadataColumn(id: Int, override val name: String, override
val dataType: DataType)
+ extends MetadataColumn {
+
+ def toPaimonDataField: DataField = {
+ new DataField(id, name, SparkTypeUtils.toPaimonType(dataType));
+ }
+}
+
+object PaimonMetadataColumn {
+
+ val FILE_PATH_COLUMN = "__paimon_file_path"
+ val ROW_INDEX_COLUMN = "__paimon_row_index"
+ val SUPPORTED_METADATA_COLUMNS: Seq[String] = Seq(FILE_PATH_COLUMN,
ROW_INDEX_COLUMN)
+
+ val ROW_INDEX: PaimonMetadataColumn =
+ PaimonMetadataColumn(Int.MaxValue - 100, ROW_INDEX_COLUMN, LongType)
+ val FILE_PATH: PaimonMetadataColumn =
+ PaimonMetadataColumn(Int.MaxValue - 101, FILE_PATH_COLUMN, StringType)
+
+ def get(metadataColumn: String): PaimonMetadataColumn = {
+ if (metadataColumn == FILE_PATH_COLUMN) {
+ FILE_PATH
+ } else if (metadataColumn == ROW_INDEX_COLUMN) {
+ ROW_INDEX
+ } else {
+ throw new IllegalArgumentException(s"$metadataColumn metadata column is
not supported.")
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
index 355902f0a..78e2e8132 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala
@@ -127,7 +127,7 @@ class PaimonMicroBatchStream(
}
override def createReaderFactory(): PartitionReaderFactory = {
- new PaimonPartitionReaderFactory(readBuilder)
+ PaimonPartitionReaderFactory(readBuilder)
}
override def initialOffset(): Offset = {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
index ef683366d..2990f8cee 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonQueryTest.scala
@@ -30,6 +30,42 @@ import scala.collection.JavaConverters._
class PaimonQueryTest extends PaimonSparkTestBase {
+ fileFormats.foreach {
+ fileFormat =>
+ bucketModes.foreach {
+ bucketMode =>
+ test(s"Query metadata columns: file.format=$fileFormat,
bucket=$bucketMode") {
+ withTable("T") {
+
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, name STRING)
+ |TBLPROPERTIES ('primary-key' = 'id',
'file.format'='$fileFormat', 'bucket'='$bucketMode')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T
+ |VALUES (1, 'x1'), (2, 'x3'), (3, 'x3'), (4, 'x4'),
(5, 'x5')
+ |""".stripMargin)
+
+ val location = loadTable("T").location().toUri.toString
+ val res = spark.sql(
+ s"""
+ |SELECT SUM(cnt)
+ |FROM (
+ | SELECT __paimon_file_path AS path, count(1) AS cnt,
count(distinct __paimon_row_index) AS dc
+ | FROM T
+ | GROUP BY __paimon_file_path
+ |)
+ |WHERE startswith(path, '$location') and endswith(path,
'.$fileFormat') and cnt == dc
+ |""".stripMargin)
+ checkAnswer(res, Row(5))
+ }
+ }
+ }
+
+ }
+
fileFormats.foreach {
fileFormat =>
bucketModes.foreach {
