xuzikun2003 commented on a change in pull request #29725: URL: https://github.com/apache/spark/pull/29725#discussion_r525900796
########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeExternalRowWindowSorter.java ########## @@ -0,0 +1,453 @@ +/* + * 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.execution; + +import com.google.common.annotations.VisibleForTesting; + +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.Map.Entry; +import java.util.Queue; +import java.util.TreeMap; +import java.io.IOException; +import java.util.function.Supplier; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import scala.collection.Iterator; +import scala.math.Ordering; + +import org.apache.spark.memory.SparkOutOfMemoryError; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.UnsafeProjection; +import org.apache.spark.sql.catalyst.expressions.UnsafeRow; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.util.collection.unsafe.sort.PrefixComparator; +import org.apache.spark.util.collection.unsafe.sort.RecordComparator; + +public final class UnsafeExternalRowWindowSorter extends AbstractUnsafeExternalRowSorter { + + private static final Logger logger = LoggerFactory.getLogger(UnsafeExternalRowWindowSorter.class); + + private final StructType schema; + private final UnsafeProjection partitionSpecProjection; + private final Ordering<InternalRow> orderingOfPartitionKey; + private final Ordering<InternalRow> orderingInWindow; + private final Ordering<InternalRow> orderingAcrossWindows; + private final PrefixComparator prefixComparatorInWindow; + private final UnsafeExternalRowSorter.PrefixComputer prefixComputerInWindow; + private final boolean canUseRadixSortInWindow; + private final long pageSizeBytes; + private static final int windowSorterMapMaxSize = 1; Review comment: In our current setting, we have one main sorter and one window sorter. If there is only window partition key on a physical partition, then all the rows will go to the window sorter and the main sorter will be empty; if there are more than one window partition key in a physical partition, then one window partition key goes to the window sorter and remaining partition keys go to the main sorter. We just reduce the original page size by half in these two sorters. We observe that halving the page size gives no performance difference in the overall TPCDS 100TB run. The advantage is that if there are very few rows inserted to the window sorter, then less memory will be wasted in the first page allocated for the window sorter and thus less overhead caused by the memory allocation of the first page. @opensky142857, You are right, it is not a rare case that one task handles several partition keys but reducing the page size by half wouldn't make much difference. The default page size is 64MB, and there is no performance difference between 64MB page size and 32 MB page size. We can also keep the page size of the main sorter unchanged. ---------------------------------------------------------------- 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: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
