You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/01/15 23:02:29 UTC
[GitHub] [iceberg] dilipbiswal opened a new pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
dilipbiswal opened a new pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095
- Creates a base trait ```RewriteRowLevelOperationHelper``` that all the rewrite rules for row level operations will make use of.
- Its a basic refactor PR which refactors the method from `RewriteDelete` rule.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi edited a comment on pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi edited a comment on pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#issuecomment-761297308
Thanks, @dilipbiswal @rdblue!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558688217
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
Review comment:
yeah, I'd also put `table` on a new line:
```
protected def buildScanPlan(
table: Table,
output: Seq[AttributeReference],
mergeBuilder: MergeBuilder,
cond: Expression): LogicalPlan = {
```
Args should be formatted with 4 spaces.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558715326
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
+
+ val scanBuilder = mergeBuilder.asScanBuilder
+
+ val predicates = splitConjunctivePredicates(cond)
+ val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
+ PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
+
+ val scan = scanBuilder.build()
+ val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
+
+ scan match {
+ case filterable: SupportsFileFilter =>
+ val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
+ val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
+ dynamicFileFilter
+ case _ =>
+ scanRelation
+ }
+ }
+
+ protected def toDataSourceFilters(predicates: Seq[Expression]): Array[sources.Filter] = {
+ predicates.flatMap { p =>
+ val translatedFilter = DataSourceStrategy.translateFilter(p, supportNestedPredicatePushdown = true)
+ if (translatedFilter.isEmpty) {
+ logWarning(s"Cannot translate expression to source filter: $p")
+ }
+ translatedFilter
+ }.toArray
+ }
+
+ protected def newWriteInfo(schema: StructType): LogicalWriteInfo = {
+ val uuid = UUID.randomUUID()
+ LogicalWriteInfoImpl(queryId = uuid.toString, schema, CaseInsensitiveStringMap.empty)
+ }
+
+ private def buildFileFilterPlan(cond: Expression, scanRelation: DataSourceV2ScanRelation): LogicalPlan = {
+ val matchingFilter = Filter(cond, scanRelation)
+ val fileAttr = findOutputAttr(matchingFilter, FILE_NAME_COL)
+ val agg = Aggregate(Seq(fileAttr), Seq(fileAttr), matchingFilter)
+ Project(Seq(findOutputAttr(agg, FILE_NAME_COL)), agg)
+ }
+
+ protected def findOutputAttr(plan: LogicalPlan, attrName: String): Attribute = {
+ val resolver = SQLConf.get.resolver
Review comment:
Can this trait define `def resolver: Resolver` so it doesn't need to use `SQLConf.get`? The resolver should be provided by the rule this is mixed into.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558682025
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDelete.scala
##########
@@ -87,35 +68,8 @@ object RewriteDelete extends Rule[LogicalPlan] with PredicateHelper with Logging
ReplaceData(r, mergeWrite, writePlan)
}
- private def buildScanPlan(
- table: Table,
- output: Seq[AttributeReference],
- mergeBuilder: MergeBuilder,
- cond: Expression): LogicalPlan = {
-
- val scanBuilder = mergeBuilder.asScanBuilder
-
- val predicates = splitConjunctivePredicates(cond)
- val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
- PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
-
- val scan = scanBuilder.build()
- val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
-
- scan match {
- case filterable: SupportsFileFilter =>
- val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
- val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
- dynamicFileFilter
- case _ =>
- scanRelation
- }
- }
-
- private def buildWritePlan(
- remainingRowsPlan: LogicalPlan,
- output: Seq[AttributeReference]): LogicalPlan = {
-
+ private def buildWritePlan(remainingRowsPlan: LogicalPlan,
+ output: Seq[AttributeReference]): LogicalPlan = {
Review comment:
Are these cosmetic changes needed?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558691370
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
Review comment:
Another nit: multiple imports on the same line cause lots of commit conflicts. Could you put these on separate lines?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] dilipbiswal commented on pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
dilipbiswal commented on pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#issuecomment-761241773
cc @aokolnychyi
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558748582
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
+
+ val scanBuilder = mergeBuilder.asScanBuilder
+
+ val predicates = splitConjunctivePredicates(cond)
+ val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
+ PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
+
+ val scan = scanBuilder.build()
+ val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
+
+ scan match {
+ case filterable: SupportsFileFilter =>
+ val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
+ val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
Review comment:
Let's just return `DynamicFileFilter(scanRelation, matchingFilePlan, filterable)` from this branch.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558748395
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
+
+ val scanBuilder = mergeBuilder.asScanBuilder
+
+ val predicates = splitConjunctivePredicates(cond)
+ val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
+ PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
+
+ val scan = scanBuilder.build()
+ val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
+
+ scan match {
+ case filterable: SupportsFileFilter =>
+ val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
+ val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
Review comment:
@dilipbiswal, can we get rid of `dynamicFileFilter` too? I know it is in the original code, I missed to update it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#issuecomment-761265505
Looks good to me apart from already pointed things.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558679350
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
Review comment:
Nit: indentation doesn't align.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558690309
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
+
+ val scanBuilder = mergeBuilder.asScanBuilder
+
+ val predicates = splitConjunctivePredicates(cond)
+ val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
+ PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
+
+ val scan = scanBuilder.build()
+ val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
+
+ scan match {
+ case filterable: SupportsFileFilter =>
+ val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
+ val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
+ dynamicFileFilter
+ case _ =>
+ scanRelation
+ }
+ }
+
+ protected def toDataSourceFilters(predicates: Seq[Expression]): Array[sources.Filter] = {
+ predicates.flatMap { p =>
+ val translatedFilter = DataSourceStrategy.translateFilter(p, supportNestedPredicatePushdown = true)
+ if (translatedFilter.isEmpty) {
+ logWarning(s"Cannot translate expression to source filter: $p")
+ }
+ translatedFilter
+ }.toArray
+ }
+
+ protected def newWriteInfo(schema: StructType): LogicalWriteInfo = {
+ val uuid = UUID.randomUUID()
+ LogicalWriteInfoImpl(queryId = uuid.toString, schema, CaseInsensitiveStringMap.empty)
+ }
+
+ private def buildFileFilterPlan(cond: Expression, scanRelation: DataSourceV2ScanRelation): LogicalPlan = {
+ val matchingFilter = Filter(cond, scanRelation)
+ val fileAttr = findOutputAttr(matchingFilter, FILE_NAME_COL)
+ val agg = Aggregate(Seq(fileAttr), Seq(fileAttr), matchingFilter)
+ Project(Seq(findOutputAttr(agg, FILE_NAME_COL)), agg)
+ }
+
+ protected def findOutputAttr(plan: LogicalPlan, attrName: String): Attribute = {
+ val resolver = SQLConf.get.resolver
+ plan.output.find(attr => resolver(attr.name, attrName)).getOrElse {
+ throw new AnalysisException(s"Cannot find $attrName in ${plan.output}")
+ }
+ }
+
+ protected def toOutputAttrs(schema: StructType, output: Seq[AttributeReference]): Seq[AttributeReference] = {
+ val nameToAttr = output.map(_.name).zip(output).toMap
+ schema.toAttributes.map {
+ a =>
Review comment:
nit: I'd prefer the original formatting where `nameToAttr.get(a.name)` is on the same line:
```
a => nameToAttr.get(a.name) match {
case Some(ref) =>
// keep the attribute id if it was present in the relation
a.withExprId(ref.exprId)
case _ =>
// if the field is new, create a new attribute
AttributeReference(a.name, a.dataType, a.nullable, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi merged pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi merged pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#issuecomment-761297308
Thanks, @dilipbiswal!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558691060
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
+
+ val scanBuilder = mergeBuilder.asScanBuilder
+
+ val predicates = splitConjunctivePredicates(cond)
+ val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
+ PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
+
+ val scan = scanBuilder.build()
+ val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
+
+ scan match {
+ case filterable: SupportsFileFilter =>
+ val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
+ val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
+ dynamicFileFilter
+ case _ =>
+ scanRelation
+ }
+ }
+
+ protected def toDataSourceFilters(predicates: Seq[Expression]): Array[sources.Filter] = {
+ predicates.flatMap { p =>
+ val translatedFilter = DataSourceStrategy.translateFilter(p, supportNestedPredicatePushdown = true)
+ if (translatedFilter.isEmpty) {
+ logWarning(s"Cannot translate expression to source filter: $p")
+ }
+ translatedFilter
+ }.toArray
+ }
+
+ protected def newWriteInfo(schema: StructType): LogicalWriteInfo = {
+ val uuid = UUID.randomUUID()
+ LogicalWriteInfoImpl(queryId = uuid.toString, schema, CaseInsensitiveStringMap.empty)
+ }
+
+ private def buildFileFilterPlan(cond: Expression, scanRelation: DataSourceV2ScanRelation): LogicalPlan = {
+ val matchingFilter = Filter(cond, scanRelation)
+ val fileAttr = findOutputAttr(matchingFilter, FILE_NAME_COL)
+ val agg = Aggregate(Seq(fileAttr), Seq(fileAttr), matchingFilter)
+ Project(Seq(findOutputAttr(agg, FILE_NAME_COL)), agg)
+ }
+
+ protected def findOutputAttr(plan: LogicalPlan, attrName: String): Attribute = {
+ val resolver = SQLConf.get.resolver
+ plan.output.find(attr => resolver(attr.name, attrName)).getOrElse {
+ throw new AnalysisException(s"Cannot find $attrName in ${plan.output}")
+ }
+ }
+
+ protected def toOutputAttrs(schema: StructType, output: Seq[AttributeReference]): Seq[AttributeReference] = {
+ val nameToAttr = output.map(_.name).zip(output).toMap
+ schema.toAttributes.map {
+ a =>
Review comment:
Or putting a => on the line above like:
```
schema.toAttributes.map { a =>
nameToAttr.get(a.name) match {
case Some(ref) =>
// keep the attribute id if it was present in the relation
a.withExprId(ref.exprId)
case _ =>
// if the field is new, create a new attribute
AttributeReference(a.name, a.dataType, a.nullable, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558688771
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.utils
+
+import java.util.UUID
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{sources, AnalysisException}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, PredicateHelper}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, DynamicFileFilter, Filter, LogicalPlan, Project}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.read.SupportsFileFilter
+import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, LogicalWriteInfoImpl}
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, PushDownUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
+ protected val FILE_NAME_COL = "_file"
+ protected val ROW_POS_COL = "_pos"
+
+ protected def buildScanPlan(table: Table,
+ output: Seq[AttributeReference],
+ mergeBuilder: MergeBuilder,
+ cond: Expression): LogicalPlan = {
Review comment:
Yeah, we should eventually make sure our style checker catches these.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #2095: Create a base trait for row level operations - Foundational work for MERGE INTO
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2095:
URL: https://github.com/apache/iceberg/pull/2095#discussion_r558685076
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDelete.scala
##########
@@ -87,35 +68,8 @@ object RewriteDelete extends Rule[LogicalPlan] with PredicateHelper with Logging
ReplaceData(r, mergeWrite, writePlan)
}
- private def buildScanPlan(
- table: Table,
- output: Seq[AttributeReference],
- mergeBuilder: MergeBuilder,
- cond: Expression): LogicalPlan = {
-
- val scanBuilder = mergeBuilder.asScanBuilder
-
- val predicates = splitConjunctivePredicates(cond)
- val normalizedPredicates = DataSourceStrategy.normalizeExprs(predicates, output)
- PushDownUtils.pushFilters(scanBuilder, normalizedPredicates)
-
- val scan = scanBuilder.build()
- val scanRelation = DataSourceV2ScanRelation(table, scan, toOutputAttrs(scan.readSchema(), output))
-
- scan match {
- case filterable: SupportsFileFilter =>
- val matchingFilePlan = buildFileFilterPlan(cond, scanRelation)
- val dynamicFileFilter = DynamicFileFilter(scanRelation, matchingFilePlan, filterable)
- dynamicFileFilter
- case _ =>
- scanRelation
- }
- }
-
- private def buildWritePlan(
- remainingRowsPlan: LogicalPlan,
- output: Seq[AttributeReference]): LogicalPlan = {
-
+ private def buildWritePlan(remainingRowsPlan: LogicalPlan,
+ output: Seq[AttributeReference]): LogicalPlan = {
Review comment:
Nit: this changes the indentation of the arguments and make it appear that this method was modified, when it wasn't. Could you revert the changes to `buildWritePlan`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org