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/23 02:40:02 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #2141: Spark: Add DistributionAndOrderingUtils
aokolnychyi opened a new pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141
This PR adds `DistributionAndOrderingUtils` that is being proposed in master and migrates `RewriteMergeInto` to use 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 a change in pull request #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564041948
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
Done. Could you check it once, @rdblue?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/DistributionAndOrderingUtils.scala
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.spark.sql.{catalyst, AnalysisException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{IcebergBucketTransform, IcebergDayTransform, IcebergHourTransform, IcebergMonthTransform, IcebergYearTransform, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, RepartitionByExpression, Sort}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{BucketTransform, DaysTransform, HoursTransform, MonthsTransform, NamedReference, Transform, YearsTransform}
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, IdentityTransform}
+import org.apache.spark.sql.connector.iceberg.distributions.{ClusteredDistribution, Distribution, OrderedDistribution, UnspecifiedDistribution}
+import org.apache.spark.sql.connector.iceberg.expressions.{NullOrdering, SortDirection, SortOrder}
+import org.apache.spark.sql.internal.SQLConf
Review comment:
Done.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564042580
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Done.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563177735
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/RewriteRowLevelOperationHelper.scala
##########
@@ -96,7 +72,7 @@ trait RewriteRowLevelOperationHelper extends PredicateHelper with Logging {
protected def buildSimpleScanPlan(
relation: DataSourceV2Relation,
- cond: catalyst.expressions.Expression): LogicalPlan = {
Review comment:
Reverted back.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563973331
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
That should work too. Will update.
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
+ return sortOrder;
+ }
+
+ Schema schema = spec.schema();
+
+ Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+
+ // build a sort prefix of partition fields that are not already in the sort order
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
Review comment:
Yeah.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Oops, I missed this one. Will revert back.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
Could you elaborate a bit more? The interface in Spark 3.2 is implemented by `Write`, not `Table`. Are you thinking of passing `SparkTable` as an arg?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
Yes, we don't want to add a round-robin repartition during inserts, for example. I'll add more info.
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Good point, let me handle this.
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
Done. Resolving this.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
Done. Could you check it once, @rdblue?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/DistributionAndOrderingUtils.scala
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.spark.sql.{catalyst, AnalysisException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{IcebergBucketTransform, IcebergDayTransform, IcebergHourTransform, IcebergMonthTransform, IcebergYearTransform, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, RepartitionByExpression, Sort}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{BucketTransform, DaysTransform, HoursTransform, MonthsTransform, NamedReference, Transform, YearsTransform}
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, IdentityTransform}
+import org.apache.spark.sql.connector.iceberg.distributions.{ClusteredDistribution, Distribution, OrderedDistribution, UnspecifiedDistribution}
+import org.apache.spark.sql.connector.iceberg.expressions.{NullOrdering, SortDirection, SortOrder}
+import org.apache.spark.sql.internal.SQLConf
Review comment:
Done.
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Added a check at the beginning of this method. Could you check, @rdblue?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Done.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
I think it looks cleaner with implicits.
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
You are right. Updated.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
I think this logic will be needed in a few places so I moved it to `Spark3Util` and got rid of implicits.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563957505
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
+ return sortOrder;
+ }
+
+ Schema schema = spec.schema();
+
+ Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+
+ // build a sort prefix of partition fields that are not already in the sort order
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
Review comment:
The logic here is unchanged from what was in `Spark3Util`, right?
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563952297
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
This is here and not in `prepareQuery` because we don't want to assume that a global ordering always requires an extra round-robin repartition? If so, it would be good to move the comment above `newChildPlan` and make it a bit more clear why this extra step is happening.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563973436
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
+ return sortOrder;
+ }
+
+ Schema schema = spec.schema();
+
+ Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+
+ // build a sort prefix of partition fields that are not already in the sort order
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
Review comment:
Yeah.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563177645
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
+ return sortOrder;
+ }
+
+ Schema schema = spec.schema();
+
+ Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+
+ // build a sort prefix of partition fields that are not already in the sort order
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
+ for (PartitionField field : spec.fields()) {
+ Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
+ boolean isSorted = sortFields.stream().anyMatch(sortField ->
+ field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
+ if (!isSorted) {
+ String sourceName = schema.findColumnName(field.sourceId());
+ builder.asc(Expressions.transform(sourceName, field.transform()));
+ }
+ }
+
+ // add the configured sort to the partition spec prefix sort
+ SortOrderVisitor.visit(sortOrder, new CopySortOrderFields(builder));
Review comment:
I removed another `CopySortOrderFields` which I think was a duplicate.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563952297
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
This is here and not in `prepareQuery` because we don't want to assume that a global ordering always requires an extra round-robin repartition?
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563973331
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
That should work too. Will update.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564131078
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
I think it looks cleaner with implicits.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564167417
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
I think this logic will be needed in a few places so I moved it to `Spark3Util` and got rid of implicits.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563986080
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Good point, let me handle this.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563949252
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
Can we expose the same interface as in Spark 3.2 from `Table` and use that here instead? Then `Spark3Util` calls remain in `SparkTable`.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Can you revert this change? I'll open a PR to update the scala style checks.
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
Should this return `SortOrder.unsorted()` instead?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
This is here and not in `prepareQuery` because we don't want to assume that a global ordering always requires an extra round-robin repartition?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
This is here and not in `prepareQuery` because we don't want to assume that a global ordering always requires an extra round-robin repartition? If so, it would be good to move the comment above `newChildPlan` and make it a bit more clear why this extra step is happening.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/DistributionAndOrderingUtils.scala
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.spark.sql.{catalyst, AnalysisException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{IcebergBucketTransform, IcebergDayTransform, IcebergHourTransform, IcebergMonthTransform, IcebergYearTransform, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, RepartitionByExpression, Sort}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{BucketTransform, DaysTransform, HoursTransform, MonthsTransform, NamedReference, Transform, YearsTransform}
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, IdentityTransform}
+import org.apache.spark.sql.connector.iceberg.distributions.{ClusteredDistribution, Distribution, OrderedDistribution, UnspecifiedDistribution}
+import org.apache.spark.sql.connector.iceberg.expressions.{NullOrdering, SortDirection, SortOrder}
+import org.apache.spark.sql.internal.SQLConf
Review comment:
Could you rewrite imports to one per line?
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Is it correct to return a clustered distribution with no expressions if the spec is unpartitioned? I think I would rather return `Distributions.unspecified` just to be safe when passing this back to Spark.
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
+ return sortOrder;
+ }
+
+ Schema schema = spec.schema();
+
+ Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
+
+ // build a sort prefix of partition fields that are not already in the sort order
+ SortOrder.Builder builder = SortOrder.builderFor(schema);
Review comment:
The logic here is unchanged from what was in `Spark3Util`, right?
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
You're right. I was thinking it was on Table instead. Let's go with this then.
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
A sorted table may not be partitioned, but it would pass the check you added. Then if the distribution mode is hash, it would return an empty clustered distribution. I think it would be more correct and easier to reason about if the check was done here.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
Should this be named `toIcebergTable` or `icebergTable`? This is doing more than just a cast, it is accessing the underlying table.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564167170
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
You are right. Updated.
----------------------------------------------------------------
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 pull request #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#issuecomment-767709184
Looks correct to me. I'd prefer to fix the unnecessary changes, but I'll leave it up to you whether to merge or fix and then merge. I usually allow nits through on the last pass to avoid blocking.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#issuecomment-767817052
Thanks for reviewing, @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] rdblue commented on a change in pull request #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563950879
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
Should this return `SortOrder.unsorted()` instead?
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564042308
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Added a check at the beginning of this method. Could you check, @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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564767850
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -52,27 +45,30 @@ import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.utils.DistributionAndOrderingUtils
import org.apache.spark.sql.catalyst.utils.PlanUtils.isIcebergRelation
import org.apache.spark.sql.catalyst.utils.RewriteRowLevelOperationHelper
import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.BooleanType
case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with RewriteRowLevelOperationHelper {
+
+ import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
+
private val ROW_FROM_SOURCE = "_row_from_source_"
private val ROW_FROM_TARGET = "_row_from_target_"
private val TRUE_LITERAL = Literal(true, BooleanType)
private val FALSE_LITERAL = Literal(false, BooleanType)
- import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
Review comment:
I'd need to check the bytecode but I agree on moving constants to the companion object. Will submit a follow-up.
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -52,27 +45,30 @@ import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.utils.DistributionAndOrderingUtils
import org.apache.spark.sql.catalyst.utils.PlanUtils.isIcebergRelation
import org.apache.spark.sql.catalyst.utils.RewriteRowLevelOperationHelper
import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.BooleanType
case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with RewriteRowLevelOperationHelper {
+
+ import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
+
private val ROW_FROM_SOURCE = "_row_from_source_"
private val ROW_FROM_TARGET = "_row_from_target_"
private val TRUE_LITERAL = Literal(true, BooleanType)
private val FALSE_LITERAL = Literal(false, BooleanType)
- import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
Review comment:
I'd need to check the bytecode but I agree on moving constants to the companion object.
Will submit a follow-up.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#issuecomment-765829839
cc @rdblue @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] rdblue commented on a change in pull request #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563957090
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
Is it correct to return a clustered distribution with no expressions if the spec is unpartitioned? I think I would rather return `Distributions.unspecified` just to be safe when passing this back to Spark.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564041653
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
Done. Resolving this.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563979984
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
+ val ordering = Spark3Util.buildRequiredOrdering(distribution, iceberg.table)
+ val newChildPlan = distribution match {
+ case _: OrderedDistribution =>
+ // insert a round robin partitioning to avoid executing the join twice
+ val numShufflePartitions = conf.numShufflePartitions
+ Repartition(numShufflePartitions, shuffle = true, childPlan)
Review comment:
Yes, we don't want to add a round-robin repartition during inserts, for example. I'll add more info.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564147727
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/IcebergImplicits.scala
##########
@@ -0,0 +1,37 @@
+/*
+ * 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 org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connector.catalog.Table
+
+object IcebergImplicits {
+ implicit class TableHelper(table: Table) {
+ def asIcebergTable: org.apache.iceberg.Table = {
Review comment:
Should this be named `toIcebergTable` or `icebergTable`? This is doing more than just a cast, it is accessing the underlying table.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi merged pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564727643
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -52,27 +45,30 @@ import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.utils.DistributionAndOrderingUtils
import org.apache.spark.sql.catalyst.utils.PlanUtils.isIcebergRelation
import org.apache.spark.sql.catalyst.utils.RewriteRowLevelOperationHelper
import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.BooleanType
case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with RewriteRowLevelOperationHelper {
+
+ import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
+
private val ROW_FROM_SOURCE = "_row_from_source_"
private val ROW_FROM_TARGET = "_row_from_target_"
private val TRUE_LITERAL = Literal(true, BooleanType)
private val FALSE_LITERAL = Literal(false, BooleanType)
- import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
Review comment:
I agree about order. We should probably also move constants into a companion class instead of inline. Does Scala do that automatically or are these initialized for every instance?
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564716971
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -52,27 +45,30 @@ import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.utils.DistributionAndOrderingUtils
import org.apache.spark.sql.catalyst.utils.PlanUtils.isIcebergRelation
import org.apache.spark.sql.catalyst.utils.RewriteRowLevelOperationHelper
import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.BooleanType
case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with RewriteRowLevelOperationHelper {
+
+ import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
+
private val ROW_FROM_SOURCE = "_row_from_source_"
private val ROW_FROM_TARGET = "_row_from_target_"
private val TRUE_LITERAL = Literal(true, BooleanType)
private val FALSE_LITERAL = Literal(false, BooleanType)
- import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
Review comment:
I think it is more natural to have imports for implicits before variables and methods in a class. I'd be in favor of changing but I can do that in a separate PR. I'll revert it from here and submit a follow-up.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563177602
##########
File path: core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.util;
+
+import java.util.Collection;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.transforms.SortOrderVisitor;
+
+public class SortOrderUtil {
+
+ private SortOrderUtil() {
+ }
+
+ public static SortOrder buildSortOrder(Table table) {
+ return buildSortOrder(table.spec(), table.sortOrder());
+ }
+
+ public static SortOrder buildSortOrder(PartitionSpec spec, SortOrder sortOrder) {
+ if (sortOrder.isUnsorted() && spec.isUnpartitioned()) {
Review comment:
I return `sortOrder` if no sort order and the spec is unpartitioned. Otherwise, we will try to build an unsorted order with id = 1, which is not allowed.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563973750
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Oops, I missed this one. Will revert back.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563953517
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/DistributionAndOrderingUtils.scala
##########
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.spark.sql.{catalyst, AnalysisException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{IcebergBucketTransform, IcebergDayTransform, IcebergHourTransform, IcebergMonthTransform, IcebergYearTransform, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, RepartitionByExpression, Sort}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.quoteIfNeeded
+import org.apache.spark.sql.connector.expressions.{BucketTransform, DaysTransform, HoursTransform, MonthsTransform, NamedReference, Transform, YearsTransform}
+import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, IdentityTransform}
+import org.apache.spark.sql.connector.iceberg.distributions.{ClusteredDistribution, Distribution, OrderedDistribution, UnspecifiedDistribution}
+import org.apache.spark.sql.connector.iceberg.expressions.{NullOrdering, SortDirection, SortOrder}
+import org.apache.spark.sql.internal.SQLConf
Review comment:
Could you rewrite imports to one per line?
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563950189
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -19,49 +19,30 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.iceberg.DistributionMode
-import org.apache.iceberg.TableProperties
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED
-import org.apache.iceberg.TableProperties.MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT
-import org.apache.iceberg.spark.Spark3Util.toClusteredDistribution
-import org.apache.iceberg.spark.Spark3Util.toOrderedDistribution
+import org.apache.iceberg.TableProperties.{MERGE_CARDINALITY_CHECK_ENABLED, MERGE_CARDINALITY_CHECK_ENABLED_DEFAULT}
+import org.apache.iceberg.spark.Spark3Util
import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.util.PropertyUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.Alias
-import org.apache.spark.sql.catalyst.expressions.Ascending
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.expressions.IsNull
import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.catalyst.expressions.NullsFirst
-import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.plans.FullOuter
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.LeftAnti
import org.apache.spark.sql.catalyst.plans.RightOuter
-import org.apache.spark.sql.catalyst.plans.logical.AppendData
-import org.apache.spark.sql.catalyst.plans.logical.DeleteAction
-import org.apache.spark.sql.catalyst.plans.logical.InsertAction
-import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.catalyst.plans.logical.JoinHint
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.MergeAction
-import org.apache.spark.sql.catalyst.plans.logical.MergeInto
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
-import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
-import org.apache.spark.sql.catalyst.plans.logical.Project
-import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
-import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
-import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, DeleteAction, InsertAction, Join, JoinHint, LogicalPlan, MergeAction, MergeInto, MergeIntoParams, MergeIntoTable, Project, Repartition, ReplaceData, UpdateAction}
Review comment:
Can you revert this change? I'll open a PR to update the scala style checks.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564702490
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -52,27 +45,30 @@ import org.apache.spark.sql.catalyst.plans.logical.MergeIntoParams
import org.apache.spark.sql.catalyst.plans.logical.MergeIntoTable
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.catalyst.plans.logical.Repartition
-import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
-import org.apache.spark.sql.catalyst.plans.logical.Sort
import org.apache.spark.sql.catalyst.plans.logical.UpdateAction
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.utils.DistributionAndOrderingUtils
import org.apache.spark.sql.catalyst.utils.PlanUtils.isIcebergRelation
import org.apache.spark.sql.catalyst.utils.RewriteRowLevelOperationHelper
import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
import org.apache.spark.sql.connector.iceberg.write.MergeBuilder
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.BooleanType
case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with RewriteRowLevelOperationHelper {
+
+ import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
+
private val ROW_FROM_SOURCE = "_row_from_source_"
private val ROW_FROM_TARGET = "_row_from_target_"
private val TRUE_LITERAL = Literal(true, BooleanType)
private val FALSE_LITERAL = Literal(false, BooleanType)
- import org.apache.spark.sql.execution.datasources.v2.ExtendedDataSourceV2Implicits._
Review comment:
Nit: this doesn't need to move the import above the constants.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563983151
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
You're right. I was thinking it was on Table instead. Let's go with this then.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r564147383
##########
File path: spark3/src/main/java/org/apache/iceberg/spark/Spark3Util.java
##########
@@ -279,46 +280,39 @@ public Transform unknown(int fieldId, String sourceName, int sourceId, String tr
return transforms.toArray(new Transform[0]);
}
- public static Distribution toOrderedDistribution(PartitionSpec spec, SortOrder sortOrder, boolean inferFromSpec) {
- if (sortOrder.isUnsorted()) {
- if (inferFromSpec) {
- SortOrder specOrder = Partitioning.sortOrderFor(spec);
- return Distributions.ordered(convert(specOrder));
- }
-
- return Distributions.unspecified();
- }
-
- Schema schema = spec.schema();
- Multimap<Integer, SortField> sortFieldIndex = Multimaps.index(sortOrder.fields(), SortField::sourceId);
-
- // build a sort prefix of partition fields that are not already in the sort order
- SortOrder.Builder builder = SortOrder.builderFor(schema);
- for (PartitionField field : spec.fields()) {
- Collection<SortField> sortFields = sortFieldIndex.get(field.sourceId());
- boolean isSorted = sortFields.stream().anyMatch(sortField ->
- field.transform().equals(sortField.transform()) || sortField.transform().satisfiesOrderOf(field.transform()));
- if (!isSorted) {
- String sourceName = schema.findColumnName(field.sourceId());
- builder.asc(org.apache.iceberg.expressions.Expressions.transform(sourceName, field.transform()));
- }
+ public static Distribution buildRequiredDistribution(org.apache.iceberg.Table table) {
+ DistributionMode distributionMode = getDistributionMode(table);
+ switch (distributionMode) {
+ case NONE:
+ return Distributions.unspecified();
+ case HASH:
+ return Distributions.clustered(toTransforms(table.spec()));
Review comment:
A sorted table may not be partitioned, but it would pass the check you added. Then if the distribution mode is hash, it would return an empty clustered distribution. I think it would be more correct and easier to reason about if the check was done here.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563949252
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
Can we expose the same interface as in Spark 3.2 from `Table` and use that here instead? Then `Spark3Util` calls remain in `SparkTable`.
----------------------------------------------------------------
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563177691
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
The intention to reuse `Spark3Util` during inserts in Spark 3.2.
----------------------------------------------------------------
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] izchen commented on a change in pull request #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
izchen commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r668627749
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/utils/DistributionAndOrderingUtils.scala
##########
@@ -0,0 +1,162 @@
+/*
+ * 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 org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.IcebergBucketTransform
+import org.apache.spark.sql.catalyst.expressions.IcebergDayTransform
+import org.apache.spark.sql.catalyst.expressions.IcebergHourTransform
+import org.apache.spark.sql.catalyst.expressions.IcebergMonthTransform
+import org.apache.spark.sql.catalyst.expressions.IcebergYearTransform
+import org.apache.spark.sql.catalyst.expressions.NamedExpression
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression
+import org.apache.spark.sql.catalyst.plans.logical.Sort
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits
+import org.apache.spark.sql.connector.expressions.BucketTransform
+import org.apache.spark.sql.connector.expressions.DaysTransform
+import org.apache.spark.sql.connector.expressions.Expression
+import org.apache.spark.sql.connector.expressions.FieldReference
+import org.apache.spark.sql.connector.expressions.HoursTransform
+import org.apache.spark.sql.connector.expressions.IdentityTransform
+import org.apache.spark.sql.connector.expressions.MonthsTransform
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.expressions.YearsTransform
+import org.apache.spark.sql.connector.iceberg.distributions.ClusteredDistribution
+import org.apache.spark.sql.connector.iceberg.distributions.Distribution
+import org.apache.spark.sql.connector.iceberg.distributions.OrderedDistribution
+import org.apache.spark.sql.connector.iceberg.distributions.UnspecifiedDistribution
+import org.apache.spark.sql.connector.iceberg.expressions.NullOrdering
+import org.apache.spark.sql.connector.iceberg.expressions.SortDirection
+import org.apache.spark.sql.connector.iceberg.expressions.SortOrder
+import org.apache.spark.sql.internal.SQLConf
+
+object DistributionAndOrderingUtils {
+
+ def prepareQuery(
+ requiredDistribution: Distribution,
+ requiredOrdering: Seq[SortOrder],
+ query: LogicalPlan,
+ conf: SQLConf): LogicalPlan = {
+
+ val resolver = conf.resolver
+
+ val distribution = requiredDistribution match {
+ case d: OrderedDistribution =>
+ d.ordering.map(e => toCatalyst(e, query, resolver))
+ case d: ClusteredDistribution =>
+ d.clustering.map(e => toCatalyst(e, query, resolver))
+ case _: UnspecifiedDistribution =>
+ Array.empty[catalyst.expressions.Expression]
+ }
+
+ val queryWithDistribution = if (distribution.nonEmpty) {
+ val numShufflePartitions = conf.numShufflePartitions
+ // the conversion to catalyst expressions above produces SortOrder expressions
+ // for OrderedDistribution and generic expressions for ClusteredDistribution
+ // this allows RepartitionByExpression to pick either range or hash partitioning
+ RepartitionByExpression(distribution, query, numShufflePartitions)
Review comment:
When `WRITE_DISTRIBUTION_MODE = range`, before this PR, Logical Plan is
```
Sort [dt#8 ASC NULLS FIRST, v#7 ASC NULLS FIRST], true
+- Repartition 2000, true
+- MergeInto
```
After this PR, Logical Plan is
```
Sort [dt#8 ASC NULLS FIRST, v#7 ASC NULLS FIRST], false
+- RepartitionByExpression [dt#8], 2000
+- Repartition 2000, true
+- MergeInto
```
In my opinion, the conversion of `global sort` to `local sort + range partitioning` is correct, but here we need to consider the `CollapseRepartition` rule in Spark `Optimizer`. In this case, this rule will eliminate the `Repartition 2000, true` node.
Please take a look here, thanks @aokolnychyi @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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2141: Spark: Add DistributionAndOrderingUtils
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #2141:
URL: https://github.com/apache/iceberg/pull/2141#discussion_r563977045
##########
File path: spark3-extensions/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteMergeInto.scala
##########
@@ -233,47 +215,23 @@ case class RewriteMergeInto(spark: SparkSession) extends Rule[LogicalPlan] with
!(actions.size == 1 && hasUnconditionalDelete(actions.headOption))
}
- def buildWritePlan(
- childPlan: LogicalPlan,
- table: Table): LogicalPlan = {
- val defaultDistributionMode = table match {
- case iceberg: SparkTable if !iceberg.table.sortOrder.isUnsorted =>
- TableProperties.WRITE_DISTRIBUTION_MODE_RANGE
- case _ =>
- TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT
- }
-
+ private def buildWritePlan(childPlan: LogicalPlan, table: Table): LogicalPlan = {
table match {
- case iceTable: SparkTable =>
- val numShufflePartitions = spark.sessionState.conf.numShufflePartitions
- val table = iceTable.table()
- val distributionMode: String = table.properties
- .getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, defaultDistributionMode)
- val order = toCatalyst(toOrderedDistribution(table.spec(), table.sortOrder(), true), childPlan)
- DistributionMode.fromName(distributionMode) match {
- case DistributionMode.NONE =>
- Sort(buildSortOrder(order), global = false, childPlan)
- case DistributionMode.HASH =>
- val clustering = toCatalyst(toClusteredDistribution(table.spec()), childPlan)
- val hashPartitioned = RepartitionByExpression(clustering, childPlan, numShufflePartitions)
- Sort(buildSortOrder(order), global = false, hashPartitioned)
- case DistributionMode.RANGE =>
- val roundRobin = Repartition(numShufflePartitions, shuffle = true, childPlan)
- Sort(buildSortOrder(order), global = true, roundRobin)
+ case iceberg: SparkTable =>
+ val distribution = Spark3Util.buildRequiredDistribution(iceberg.table)
Review comment:
Could you elaborate a bit more? The interface in Spark 3.2 is implemented by `Write`, not `Table`. Are you thinking of passing `SparkTable` as an arg?
----------------------------------------------------------------
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